Hash 分区 AttributeMap

当记录有稳定的查找 key，而且每条记录都是结构化数据时，使用这个模式。示例按 email 地址保存客户 profile。一次查找只读取一个 partition，而不是整个目录。

CustomerDirectoryFlow 是只处理 RPC 的 Flow。只需启动一次，不设置 Flow timeout。CustomerProfilesByEmailPartition 这个 AttributeMap 最多有 1,000 个 instance，每个 instance 在需要时才出现。每个 instance 保存一个从 canonical email 到完整 profile 的 dictionary。

稳定分区算法

每种 SDK 使用完全相同的算法：

Email 地址必须是非空 ASCII。
去除首尾的 ASCII 空白。
把 ASCII 字母 A 到 Z 转为小写。
对 canonical email 的字节计算 32-bit FNV-1a，溢出时回绕。
计算 hash modulo 1000，把 instance 名格式化为 partition-000 到 partition-999。

Hash 冲突只会让多个 email 落进同一个 partition。Dictionary 仍然用完整的 canonical email 区分记录。

精确查找与串行写入

Controller 和 RPC handler 各自计算 partition。Controller 调用 UpsertCustomerProfile 时，要求 Dex 只加载这一个 partition instance 并给它加锁。Handler 再次计算 partition，读取 bucket，更新一条 profile，再写回 bucket。同一 partition 的写入逐个执行，不同 partition 可以并发写入。

同一 partition 的并发写入可能收到 lock-conflict 错误。upsert 是幂等的，所以直接重试，见 RPC lock conflicts。要协调整张 map 的写入，见整张 map 的协调锁。

GetCustomerProfileByEmail 只加载一个 partition，不增加写锁。Partition 不存在或 canonical email 不存在时都会返回 not found。

controller.pyPython
12345678910111213
saved_profile = await app_state.client.invoke_rpc(
    flow.upsert_customer_profile,
    FLOW_ID,
    profile,
    options=RPCInvokeOptions(
        lock_attribute_map_instances=(
            flow.customer_profiles_by_email_partition.lock(partition_name),
        ),
        load_attribute_map_instances=(
            flow.customer_profiles_by_email_partition.load(partition_name),
        ),
    ),
)

例子: examples/python/dex_examples/patterns/hash-partitioned-attribute-map/controller.py

controller.goGo
123456789101112131415
err = controller.client.InvokeRPCWithOptions(
	request.Request.Context(),
	FlowID,
	controller.flow.UpsertCustomerProfile,
	profile,
	&savedProfile,
	dex.RPCInvokeOptions{
		LockAttributeMapInstances: []dex.AttributeLock{
			dex.LockAttributeMap(CustomerProfilesByEmailPartition, partitionName),
		},
		LoadAttributeMapInstances: []dex.AttributeMapLoad{
			CustomerProfilesByEmailPartition.Load(partitionName),
		},
	},
)

例子: examples/go/patterns/hash-partitioned-attribute-map/controller.go

HashPartitionedAttributeMapController.javaJava
123456789101112
final RPCInvokeOptions options = RPCInvokeOptions.newBuilder()
        .addLockAttributeMapInstance(
                flow.customerProfilesByEmailPartition,
                partition.partitionName)
        .addLoadAttributeMapInstance(
                flow.customerProfilesByEmailPartition,
                partition.partitionName)
        .build();
return ResponseEntity.ok(client.invokeRPC(
        stub::upsertCustomerProfile,
        profile,
        options));

例子: examples/java/src/main/java/io/superdurable/dex/patterns/hashpartitionedattributemap/HashPartitionedAttributeMapController.java

controller.tsTypeScript
123456789
await client.invokeRPCWithOptions(
  customerDirectoryFlow.upsertCustomerProfile,
  CUSTOMER_DIRECTORY_FLOW_ID,
  profile,
  {
    lockAttributeMapInstances: [profiles.lock(partitionName)],
    loadAttributeMapInstances: [profiles.load(partitionName)],
  },
)

例子: examples/typescript/src/patterns/hash-partitioned-attribute-map/controller.ts

controller.rsRust
123456789101112
client.invoke_rpc_with_options(
    FLOW_ID,
    UPSERT_CUSTOMER_PROFILE,
    profile,
    RpcInvokeOptions::new()
        .lock_attribute_map_instance(
            CUSTOMER_PROFILES_BY_EMAIL_PARTITION.lock(&partition.partition_name),
        )
        .load_attribute_map_instance(
            CUSTOMER_PROFILES_BY_EMAIL_PARTITION.load(partition.partition_name),
        ),
)

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

每个写入方都必须给它要写的 partition instance 加锁。锁只能挡住同样加锁的其他写入方。不加锁的写入方仍可能覆盖它们的修改。

HTTP API

POST /patterns/hash-partitioned-attribute-map/start
PUT  /patterns/hash-partitioned-attribute-map/customer-profile
GET  /patterns/hash-partitioned-attribute-map/customer-profile?emailAddress=alice@example.com

Partition 数量是持久化数据的一部分。修改它会把大多数 key 移到别的 partition，因此需要完整 rehash 或新的 Flow 版本。这个示例没有实现删除、全目录枚举、跨字段搜索和调整 partition 数量。

下一步

带 SQL 投影的实体 Flow：为每个实体（例如一个用户）保留一个 Flow。