Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions .specgit.yaml
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
version: 1
delivery: summary-diff-continue
delivery: event-idempotency-gate
context:
kind: branch
branch: fix/525-summary-diff-continue
branch: fix/523-event-idempotency-gate
issues:
- 525
pr: 526
- 523
pr: 527
14 changes: 12 additions & 2 deletions packages/core/schema.json
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
{
"version": "7",
"dialect": "sqlite",
"id": "abadf28b-1770-46c6-bbf6-b7800a9ca874",
"id": "7a2e2a70-584a-4604-bf73-4c6e116c20a3",
"prevIds": [
"874d8e74-d354-4dcb-b98c-c893660c9371"
"abadf28b-1770-46c6-bbf6-b7800a9ca874"
],
"ddl": [
{
Expand Down Expand Up @@ -1052,6 +1052,16 @@
"entityType": "columns",
"table": "event"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "data_hash",
"entityType": "columns",
"table": "event"
},
{
"type": "text",
"notNull": false,
Expand Down
1 change: 1 addition & 0 deletions packages/core/src/database/migration.gen.ts

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"

export default {
id: "20260903062324_add_event_data_hash",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`ALTER TABLE \`event\` ADD \`data_hash\` text;`)
})
},
} satisfies DatabaseMigration.Migration
1 change: 1 addition & 0 deletions packages/core/src/database/schema.gen.ts
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,7 @@ export default {
\`seq\` integer NOT NULL,
\`type\` text NOT NULL,
\`data\` text NOT NULL,
\`data_hash\` text,
CONSTRAINT \`fk_event_aggregate_id_event_sequence_aggregate_id_fk\` FOREIGN KEY (\`aggregate_id\`) REFERENCES \`event_sequence\`(\`aggregate_id\`) ON DELETE CASCADE
);
`)
Expand Down
58 changes: 43 additions & 15 deletions packages/core/src/event.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,14 @@ export * as EventV2 from "./event"
import { Cause, Context, Effect, FiberSet, Layer, Option, PubSub, Schema, Stream } from "effect"
import { Event } from "@opencode-ai/schema/event"
import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
import { and, asc, eq, gt } from "drizzle-orm"
import { and, asc, desc, eq, gt } from "drizzle-orm"
import { Database } from "./database/database"
import { EventSequenceTable, EventTable } from "./event/sql"
import { Location } from "./location"
import { LayerNode } from "./effect/layer-node"
import { isDeepStrictEqual } from "node:util"
import { Durable } from "@opencode-ai/schema/durable-event-manifest"
import { Hash } from "./util/hash"

export const ID = Event.ID
export type ID = import("@opencode-ai/schema/event").ID
Expand Down Expand Up @@ -227,6 +228,30 @@ export const layerWith = (options?: LayerOptions) =>
if (input && row?.ownerID && row.ownerID !== input.ownerID) {
return undefined
}
const dataHash = Hash.sha256(JSON.stringify(encoded))
if (!input) {
// Idempotency gate (#523): a fresh append that byte-for-byte repeats
// this aggregate's latest same-type event carries zero information
// delta. Skip it entirely — no seq consumed, no projectors, no commit
// hook, no durable wake — so the persisted sequence stays dense and
// both the replayAll contiguity check and gt(seq, after) readers are
// unaffected. Replay appends (input) keep their exact-seq contract,
// and legacy rows carry a NULL hash so they never match.
const previous = yield* db
.select({ dataHash: EventTable.data_hash })
.from(EventTable)
.where(
and(
eq(EventTable.aggregate_id, aggregateID),
eq(EventTable.type, versionedType(definition.type, durable.version)),
),
)
.orderBy(desc(EventTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
if (previous && previous.dataHash === dataHash) return undefined
}
const seq = input?.seq ?? latest + 1
if (input && seq !== latest + 1) {
yield* Effect.die(
Expand Down Expand Up @@ -278,6 +303,7 @@ export const layerWith = (options?: LayerOptions) =>
seq,
type: versionedType(definition.type, durable.version),
data: encoded,
data_hash: dataHash,
},
])
.run()
Expand Down Expand Up @@ -479,7 +505,9 @@ export const layerWith = (options?: LayerOptions) =>
.transaction(
() =>
Effect.gen(function* () {
const results = new Array<{ aggregateID: string; seq: number }>()
// Aligned with entries by index: a deduped entry yields
// undefined so the payload pairing below stays positional.
const results = new Array<{ aggregateID: string; seq: number } | undefined>()
for (const entry of entries) {
// No replay input: seq is allocated contiguously from the latest sequence inside the transaction.
const result = yield* commitDurableEventInner(
Expand All @@ -488,7 +516,7 @@ export const layerWith = (options?: LayerOptions) =>
undefined,
entry.commit,
)
if (result) results.push(result)
results.push(result)
}
return results
}),
Expand All @@ -499,19 +527,19 @@ export const layerWith = (options?: LayerOptions) =>
return results
}),
)
const payloads = entries.flatMap((entry, index) => {
const payloads = entries.map((entry, index) => {
const result = committed[index]
if (!result) return []
return [
{
...entry.event,
durable: {
aggregateID: result.aggregateID,
seq: result.seq,
version: entry.durable.version,
},
} as Payload,
]
// A deduped entry is still notified (mirrors the single-publish
// path) but stays unstamped: it occupies no sequence position.
if (!result) return entry.event
return {
...entry.event,
durable: {
aggregateID: result.aggregateID,
seq: result.seq,
version: entry.durable.version,
},
} as Payload
})
for (const payload of payloads) {
yield* notify(payload)
Expand Down
5 changes: 5 additions & 0 deletions packages/core/src/event/sql.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@ export const EventTable = sqliteTable(
seq: integer().notNull(),
type: text().notNull(),
data: text({ mode: "json" }).$type<Record<string, unknown>>().notNull(),
// sha256 of the serialized payload, written once at append time. The
// idempotency gate compares against the latest same-type row via
// event_aggregate_type_seq_idx instead of re-hashing MiB-scale payloads.
// Nullable: legacy rows predate the column and never match the gate.
data_hash: text(),
},
(table) => [
uniqueIndex("event_aggregate_seq_idx").on(table.aggregate_id, table.seq),
Expand Down
Loading
Loading