Add the table name and the target ID to each push batch

docs/agent-plans/2026-10-17-coalesced-materialization-and-gsis.md

@@ -284,8 +284,8 @@ Materialization Shard Group: fokos.m.<tableName> (MaterializationDO)
   - Receiver: producer_fences; in-memory runner round-robins targets (2 workers); alarm is fallback
           |
           +-----------------------------------------------+
-          | gsiWriteEntries(producerId, epoch,            | PushTargetBatch(producerId, epoch,
-          |   ancestorFences, entries)                    |   ancestorFences, items)
+          | gsiWriteEntries(producerId, epoch,            | PushTargetBatch(tableName, targetId,
+          |   ancestorFences, entries)                    |   producerId, epoch, ancestorFences, items)
           v                                               v
 GSI Shard Groups (up to 10):                       Remote Push Targets:
   fokos.gsiv1.<n>.<tableName>.<indexName>            Workers RPC to a WorkerEntrypoint or DurableObject
@@ -859,6 +859,10 @@ export type PushTargetItem = {
 };
 
 export type PushTargetBatch = {
+	/** The name of the base table, with no `fokos.` shard group prefix. */
+	tableName: string;
+	/** The target that sends the batch. Each target has its own epoch and its own ancestor fences. */
+	targetId: string;
 	/** The `DurableObjectId` of the sending `MaterializationDO`, as `DurableObjectId.toString()` gives it (hex). */
 	producerId: string;
 	epoch: number;
@@ -884,7 +888,13 @@ The push target contract:
 - `revision` strictly increases for each `(baseHashKey, baseSortKey)` across creates, updates, and deletes.
 - Two pushes for one key can arrive out of order when an earlier call stays in flight after a lost response. Thus the
   push target must keep the highest applied `revision` of each key and ignore each push with
-  `revision <= appliedRevision`.
+  `revision <= appliedRevision`. The key is `(tableName, baseHashKey, baseSortKey)`, because one push target can
+  receive the batches of more than one table.
+- A push target that checks the producer epoch uses the receiver rules of Section 4.2.9 with the producer
+  `(producerId, targetId)` and the ancestor `(ancestorId, targetId)`. Each target of one `MaterializationDO` has its
+  own epoch, and two targets of one table can send to the same push target.
+- The batch carries no other part of the table configuration. The policy holds the bindings and the split settings
+  of the producer, and a push target has no use for them.
 - A push target that checks the producer epoch with the receiver rules of Section 4.2.9 can delete the revision of a
   deleted key after it applies the delete. A push target that does not check the epoch must keep that revision. Its
   retention is the choice of the push target.
@@ -897,7 +907,7 @@ The push target contract:
 | --- | --- | --- | --- |
 | Base `PartitionDO` -> `MaterializationDO` (`applyLatestImages`) | `DurableObjectId` of the base partition, in hex | `materialization_export_state` | The producer epoch, then the revision (Section 4.2.4) |
 | `MaterializationDO` -> GSI `PartitionDO` (`gsiWriteEntries`) | `DurableObjectId` of the `MaterializationDO`, in hex | `materialization_targets.epoch` of the GSI target | The producer epoch |
-| `MaterializationDO` -> push target | `DurableObjectId` of the `MaterializationDO`, in hex | `materialization_targets.epoch` of the push target | The revision. The producer epoch is optional (Section 4.2.8). |
+| `MaterializationDO` -> push target | `DurableObjectId` of the `MaterializationDO`, in hex, with `targetId` | `materialization_targets.epoch` of the push target | The revision. The producer epoch is optional (Section 4.2.8). |
 
 Each GSI shard group gets calls from one target only. Thus a GSI partition stores one epoch for each
 `MaterializationDO`, and the producer ID needs no `targetId`. The producer ID is the `DurableObjectId` as

packages/fokosdb/src/server/materialization/do-materialization.ts

@@ -1,5 +1,5 @@
 import { DurableObject } from "cloudflare:workers";
-import type { FokosDBPolicy, FokosDBRouteContext } from "../../shared/partition-context.js";
+import { baseTableName, type FokosDBPolicy, type FokosDBRouteContext } from "../../shared/partition-context.js";
 import { FokosShardingRuntime } from "../../sharding/runtime.js";
 import type { FokosRuntimeConfigOverrides } from "../../sharding/runtime-config.js";
 import type { FokosEnvelope, FokosLifecycle, FokosLocalCall, FokosOperations, FokosShardingHooks } from "../../sharding/runtime-types.js";
@@ -151,6 +151,7 @@ export class FokosDBMaterializationDO extends DurableObject<Env> implements Mate
 				const owns = this.fokos.ownerCheck();
 				return (row) => owns({ hashKey: row.hk, sortKey: row.sk });
 			},
+			tableName: () => baseTableName(this.fokos.identity().topology),
 			producerId: () => this.ctx.id.toString(),
 			checkPushTarget: (target) => {
 				pushTargetStub(this.env, target);

packages/fokosdb/src/server/materialization/target-runner.test.ts

@@ -101,6 +101,7 @@ async function withRunner(
 			targets: () => targets,
 			canRun: () => true,
 			ownerCheck: () => (row) => opts.owns?.(KeyCodec.decode(row.hk) as string) ?? true,
+			tableName: () => "orders",
 			producerId: () => "aa".repeat(32),
 			checkPushTarget: (target) => opts.checkPushTarget?.(target.targetId),
 			push: async (target, batch) => {
@@ -189,6 +190,8 @@ describe("TargetRunner — batches and cursors", () => {
 
 			expect(h.calls).toHaveLength(1);
 			expect(h.calls[0].batch).toEqual({
+				tableName: "orders",
+				targetId: "t1",
 				producerId: "aa".repeat(32),
 				epoch: 1,
 				ancestorFences: [],

packages/fokosdb/src/server/materialization/target-runner.ts

@@ -93,6 +93,8 @@ export type TargetRunnerDeps = {
 	canRun: () => boolean;
 	/** A check of the keys that this partition owns now. The runner gets a new check for each transaction. */
 	ownerCheck: () => (row: { hk: KeyBytes; sk: KeyBytes }) => boolean;
+	/** The name of the base table. */
+	tableName: () => string;
 	/** The `DurableObjectId` of this partition, in hex. */
 	producerId: () => string;
 	/**
@@ -719,6 +721,8 @@ export class TargetRunner {
 	 */
 	async #send(target: PushSinkTargetDefinition, batch: ClaimedBatch): Promise<PushTargetResult[] | null> {
 		const request: PushTargetBatch = {
+			tableName: this.#deps.tableName(),
+			targetId: target.targetId,
 			producerId: this.#deps.producerId(),
 			epoch: batch.epoch,
 			ancestorFences: batch.ancestorFences,

packages/fokosdb/src/server/materialization/targets.ts

@@ -40,6 +40,10 @@ export type PushTargetItem = {
 };
 
 export type PushTargetBatch = {
+	/** The name of the base table, with no `fokos.` shard group prefix. */
+	tableName: string;
+	/** The target that sends the batch. Each target has its own epoch and its own ancestor fences. */
+	targetId: string;
 	/** The `DurableObjectId` of the sending `FokosDBMaterializationDO`, as `DurableObjectId.toString()` gives it (hex). */
 	producerId: string;
 	epoch: number;

packages/fokosdb/src/shared/partition-context.ts

@@ -110,9 +110,7 @@ export function materializationTableConfig(table: FokosDBTableConfig): FokosDBTa
  * gives one shard group. The name is the address of the GSI data, so it must never change.
  */
 export function gsiShardGroup(topology: FokosTopology, indexName: string): string {
-	const tableName = topology.shardGroup.startsWith(MATERIALIZATION_SHARD_GROUP_PREFIX)
-		? topology.shardGroup.slice(MATERIALIZATION_SHARD_GROUP_PREFIX.length)
-		: tableNameOf(topology);
+	const tableName = baseTableName(topology);
 	return `${RESERVED_SHARD_GROUP_PREFIX}gsiv1.${tableName.length}.${tableName}.${indexName}`;
 }
 
@@ -146,6 +144,13 @@ export function gsiTableConfig(table: FokosDBTableConfig, indexName: string): Fo
 	};
 }
 
+/** The table name, from the partition group or the materialization group of the table. */
+export function baseTableName(topology: FokosTopology): string {
+	return topology.shardGroup.startsWith(MATERIALIZATION_SHARD_GROUP_PREFIX)
+		? topology.shardGroup.slice(MATERIALIZATION_SHARD_GROUP_PREFIX.length)
+		: tableNameOf(topology);
+}
+
 function tableNameOf(topology: FokosTopology): string {
 	invariant(
 		topology.shardGroup.startsWith(PARTITION_SHARD_GROUP_PREFIX),

packages/fokosdb/test/materialization-do/push-target-host.ts

@@ -17,7 +17,9 @@ export class PushSinkDO extends DurableObject<Env> {
 			`CREATE TABLE IF NOT EXISTS received (producer TEXT NOT NULL, hk TEXT NOT NULL, revision INTEGER NOT NULL, deleted INTEGER NOT NULL, data TEXT, epoch INTEGER NOT NULL)`,
 		);
 		ctx.storage.sql.exec(`CREATE TABLE IF NOT EXISTS refused (producer TEXT PRIMARY KEY)`);
-		ctx.storage.sql.exec(`CREATE TABLE IF NOT EXISTS calls (producer TEXT NOT NULL, fences TEXT NOT NULL)`);
+		ctx.storage.sql.exec(
+			`CREATE TABLE IF NOT EXISTS calls (producer TEXT NOT NULL, table_name TEXT NOT NULL, target_id TEXT NOT NULL, fences TEXT NOT NULL)`,
+		);
 	}
 
 	/** Makes each later call of `producerId` fail as a whole, so the cursor of its target stays. */
@@ -26,7 +28,13 @@ export class PushSinkDO extends DurableObject<Env> {
 	}
 
 	push(batch: PushTargetBatch): PushTargetResult[] {
-		this.ctx.storage.sql.exec(`INSERT INTO calls (producer, fences) VALUES (?, ?)`, batch.producerId, JSON.stringify(batch.ancestorFences));
+		this.ctx.storage.sql.exec(
+			`INSERT INTO calls (producer, table_name, target_id, fences) VALUES (?, ?, ?, ?)`,
+			batch.producerId,
+			batch.tableName,
+			batch.targetId,
+			JSON.stringify(batch.ancestorFences),
+		);
 		if (this.ctx.storage.sql.exec(`SELECT 1 FROM refused WHERE producer = ?`, batch.producerId).toArray().length > 0) {
 			return batch.items.map(() => ({ outcome: "retry", code: "refused" }));
 		}
@@ -45,6 +53,17 @@ export class PushSinkDO extends DurableObject<Env> {
 		});
 	}
 
+	/** The table and the target of each call of `producerId`, in call order. */
+	senders(producerId: string): Pick<PushTargetBatch, "tableName" | "targetId">[] {
+		return this.ctx.storage.sql
+			.exec<{ table_name: string; target_id: string }>(
+				`SELECT table_name, target_id FROM calls WHERE producer = ? ORDER BY rowid`,
+				producerId,
+			)
+			.toArray()
+			.map((row) => ({ tableName: row.table_name, targetId: row.target_id }));
+	}
+
 	/** The ancestor fences of each call of `producerId`, in call order. */
 	fences(producerId: string): PushTargetBatch["ancestorFences"][] {
 		return this.ctx.storage.sql

packages/fokosdb/test/materialization-do/push-targets.test.ts

@@ -50,6 +50,11 @@ describe("FokosDBMaterializationDO — push targets", () => {
 
 		// The target read the tombstone, so the cleanup deletes it and each position.
 		await vi.waitFor(async () => expect(await materializationState(mat)).toMatchObject({ items: 0, positions: 0 }), TARGET_WAIT);
+		// Each call names the base table without the shard group prefix, and the target that sent it.
+		const tableName = ctx.topology.shardGroup.slice("fokos.p.".length);
+		const senders = await env.PUSH_SINK_DO.getByName("sink").senders(producerId);
+		expect(senders.length).toBeGreaterThan(0);
+		expect(senders).toEqual(senders.map(() => ({ tableName, targetId: "sink" })));
 		expect(await materializationState(mat)).toMatchObject({
 			targets: [{ target_id: "sink", kind: "push_sink", epoch: 1, is_batch_open: 0 }],
 			storedTargets: [{ kind: "push_sink", targetId: "sink", binding: "PUSH_SINK_DO", method: "push", objectName: "sink" }],