构建时序数据库聚合器

项目概述

本项目实现一个面向 time-series data aggregation 的完整系统，核心要解决的问题是 compressing high-frequency samples into queryable buckets without losing p99 fidelity。整体由 Downsample buckets、Retention policy、Compression encoder、Query rollup、Ingest backpressure 五个模块组成，每个模块单一职责、可独立测试，最后通过集成测试验证端到端正确性。设计上把 a write burst that arrives out of order and breaks the monotonically-increasing timestamp assumption 作为第一类需求先行处理，而非事后再补。

架构与关键决策

数据模型为 a column of timestamps paired with scalar values, bucketed by a configurable interval and stored as delta-of-delta encoded blocks。Downsample buckets 定义其余阶段共同依赖的数据形状并在边界处校验输入；Retention policy 承载 time-series data aggregation 的正确性逻辑；Compression encoder 专门处理 a write burst that arrives out of order and breaks the monotonically-increasing timestamp assumption，在失败路径上给出明确的拒绝、重试或降级策略；Query rollup 提供可观测性，每次操作输出带关联 id 的结构化事件；Ingest backpressure 在过载时按既定方式降级，避免过载演变为数据损坏。配置与代码分离，使风险变更可经 diff 审阅。

实施与测试

按五个阶段依次实现 Downsample buckets、Retention policy、Compression encoder、Query rollup、Ingest backpressure，每阶段配套聚焦单测。集成测试组合各模块并驱动真实 time-series data aggregation 场景：先验证正常路径，再刻意触发 a write burst that arrives out of order and breaks the monotonically-increasing timestamp assumption 并断言系统降级而非损坏。所有依赖均在进程内或被桩替换，用例在 CI 中确定可复现。

限制与后续

• 单区域设计，多区域复制超出范围。
• Downsample buckets 默认进程内，未水平共享。
• 无鉴权层，假定部署在强制身份的边缘之后。

代码示例

# setup for time-series-db-aggregator
pnpm init
pnpm add express lmdb msgpack
pnpm add -D typescript tsx vitest

time_series_db_aggregator/
├── src/index.ts
├── src/downsample-buckets.ts
├── src/retention-policy.ts
├── tests/time_series_db_aggregator.test.ts
└── package.json

// Downsample buckets: time-series data aggregation core
export function downsamplebuckets(input: DownsamplebucketsInput): DownsamplebucketsResult {
  // Build the downsample buckets that anchors time-series data aggregation and defines the core data shape.
  return processDownsamplebuckets(input);
}

// Retention policy — handles time-series data aggregation Add the retention policy so
export class Retentionpolicy {
  constructor(private opts: RetentionpolicyOptions) {}
  async run(ctx: Context): Promise<RetentionpolicyOutput> {
    return this.execute(ctx);
  }
}

// Compression encoder: time-series data aggregation stage
export const compressionencoder = (next: Next) => async (req: Request) => {
  const out = await transform(req);
  return next(out);
};

// Query rollup for time-series data aggregation
export async function queryrollup(stream: AsyncIterable<Chunk>): Promise<Aggregate> {
  let acc = init();
  for await (const c of stream) acc = merge(acc, c);
  return acc;
}

// Ingest backpressure: time-series data aggregation finalization
export function ingestbackpressure(state: State): Result {
  validate(state);
  return finalize(state);
}

# run the time-series data aggregation test suite
pnpm vitest run