顺序分块 AttributeMap

查看 Markdown查看 Markdown

这个模式适合需要保留写入顺序的 append-only 数据，例如 Agent messages、subscriber history 或 audit events。

如果把这些数据放在单个 Attribute 里，它的 value 会越来越大，每次 append 都要读取并重写整个 value。顺序分块的 AttributeMap 把数据拆成多个 chunk，每次 append 只重写一个大小有上限的 chunk。示例的每个 instance 最多保存 100 个 subscriber。

数据布局

ChunkedSubscriberFlow 是 RPC-only Flow。它只需手动启动一次，不设置 Flow timeout。

• SubscriberArchiveState 保存下一个 sequence 和最新 archive chunk token。
• SubscriberChunks[current] 是唯一可修改的 chunk。
• Chunk 满后会写入不可变 instance，名称是第一条 sequence 的 20 位补零形式，例如 00000000000000000101。

RegisterSubscriber 会锁定 state 和 current instance。大多数调用只重写 current。如果 current 已有 100 条记录，同一次 RPC 调用会写入 archive chunk、创建新的 current，并推进 state。这些写入一起生效。

并发 append 可能收到 lock-conflict 错误。收到后重试，见 RPC lock conflicts。要协调整张 map 的写入，见 整张 map 的协调锁。

GetSubscriberPage 默认从 current 开始，也可以接收 page token。HTTP controller 使用 RPCInvokeOptions 只加载目标 instance。每页的记录按从新到旧排列，并带上更旧一页的 token。

controller.pyPython

12345678

page = await app_state.client.invoke_rpc(
    flow.get_subscriber_page,
    FLOW_ID,
    GetSubscriberPageInput(page_token),
    options=RPCInvokeOptions(
        load_attribute_map_instances=(flow.subscriber_chunks.load(page_token),)
    ),
)

例子: examples/python/dex_examples/patterns/sequentially-chunked-attribute-map/controller.py

controller.goGo

123456789101112

err = controller.client.InvokeRPCWithOptions(
	request.Request.Context(),
	FlowID,
	controller.flow.GetSubscriberPage,
	GetSubscriberPageInput{PageToken: pageToken},
	&page,
	dex.RPCInvokeOptions{
		LoadAttributeMapInstances: []dex.AttributeMapLoad{
			SubscriberChunks.Load(pageToken),
		},
	},
)

例子: examples/go/patterns/sequentially-chunked-attribute-map/controller.go

SequentiallyChunkedAttributeMapController.javaJava

12345678910

final RPCInvokeOptions options = RPCInvokeOptions.newBuilder()
        .addLoadAttributeMapInstance(flow.subscriberChunks, validatedToken)
        .build();
final ChunkedSubscriberFlow stub = client.newRpcStub(
        ChunkedSubscriberFlow.class,
        ChunkedSubscriberFlow.FLOW_ID);
return ResponseEntity.ok(client.invokeRPC(
        stub::getSubscriberPage,
        new ChunkedSubscriberFlow.GetSubscriberPageInput(validatedToken),
        options));

例子: examples/java/src/main/java/io/superdurable/dex/patterns/sequentiallychunkedattributemap/SequentiallyChunkedAttributeMapController.java

controller.tsTypeScript

12345678910

await client.invokeRPCWithOptions(
  chunkedSubscriberFlow.getSubscriberPage,
  CHUNKED_SUBSCRIBER_FLOW_ID,
  { pageToken },
  {
    loadAttributeMapInstances: [
      chunkedSubscriberFlow.subscriberChunks.load(pageToken),
    ],
  },
)

例子: examples/typescript/src/patterns/sequentially-chunked-attribute-map/controller.ts

controller.rsRust

123456789

client.invoke_rpc_with_options(
    FLOW_ID,
    GET_SUBSCRIBER_PAGE,
    GetSubscriberPageInput {
        page_token: page_token.clone(),
    },
    RpcInvokeOptions::new()
        .load_attribute_map_instance(SUBSCRIBER_CHUNKS.load(page_token)),
)

例子: examples/rust/src/patterns/sequentially_chunked_attribute_map/controller.rs

分页规则

Flow 启动后，current token 始终有效。空的 current 返回空页。Archive token 必须是 20 位十进制数字，并且指向 chunk 边界。格式错误的 token，或格式正确但没有对应 chunk 的 token，都会返回明确错误。

Archive chunk 不会再改变，所以同一页再读一次（例如重试时）得到的记录相同。如果需要更新旧记录，请换用其他模式或增加独立索引。

HTTP API

POST /patterns/sequentially-chunked-attribute-map/start
POST /patterns/sequentially-chunked-attribute-map/register
GET  /patterns/sequentially-chunked-attribute-map/subscribers?pageToken=current

chunk size 是存储布局的一部分。应用的新版本可以为之后的 archive 使用新的 chunk size，但它的 page token 逻辑必须能读取两种布局。

下一步

Hash 分区 AttributeMap