diff --git a/__tests__/database-writer-pool-concurrency.test.ts b/__tests__/database-writer-pool-concurrency.test.ts
new file mode 100644
index 0000000..45654e3
--- /dev/null
+++ b/__tests__/database-writer-pool-concurrency.test.ts
@@ -0,0 +1,552 @@
+import Database from "better-sqlite3";
+import {
+  setDb,
+  runMigrations,
+  closeDb,
+  insertEvent,
+  type EventRow,
+} from "../src/indexer/db.js";
+import {
+  WriterPoolEventQueue,
+  WriterPoolEventQueueOverflowError,
+  DEFAULT_WRITER_POOL_EVENT_QUEUE_MAX_SIZE,
+  writerPoolEventIdentityKey,
+  submitEventNotifications,
+  getWriterPoolEventQueue,
+  resetWriterPoolStartState,
+  queueWrite,
+  flushWriteQueue,
+  type WriteOperation,
+} from "../src/indexer/database-writer-pool.js";
+
+const CONTRACT_ID = "CTEST0000000000000000000000000000000000000000000000000001";
+const EVENT_TYPES = ["initialized", "funded", "approved"];
+
+function row(
+  ledgerSequence: number,
+  eventType = "funded",
+  contractId = CONTRACT_ID,
+): EventRow {
+  return {
+    contractId,
+    eventType,
+    ledgerSequence,
+    timestamp: 1_700_000_000 + ledgerSequence,
+    dataJson: JSON.stringify({ ledger: ledgerSequence, eventType }),
+  };
+}
+
+function rows(start: number, end: number, perLedger = 1): EventRow[] {
+  const out: EventRow[] = [];
+  for (let ledger = start; ledger <= end; ledger++) {
+    for (let i = 0; i < perLedger; i++) {
+      out.push(row(ledger, EVENT_TYPES[i % EVENT_TYPES.length]));
+    }
+  }
+  return out;
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+describe("database_writer_pool – concurrent event insert locks (#327)", () => {
+  afterEach(async () => {
+    await flushWriteQueue();
+    resetWriterPoolStartState();
+  });
+
+  describe("writerPoolEventIdentityKey", () => {
+    it("keys on the events table's uniqueness triple", () => {
+      expect(writerPoolEventIdentityKey(row(7, "funded"))).toBe(
+        `${CONTRACT_ID}|7|funded`,
+      );
+    });
+
+    it("distinguishes ledger, type and contract", () => {
+      const base = writerPoolEventIdentityKey(row(7, "funded"));
+      expect(writerPoolEventIdentityKey(row(8, "funded"))).not.toBe(base);
+      expect(writerPoolEventIdentityKey(row(7, "approved"))).not.toBe(base);
+      expect(writerPoolEventIdentityKey(row(7, "funded", "COTHER"))).not.toBe(
+        base,
+      );
+    });
+  });
+
+  describe("WriterPoolEventQueue", () => {
+    it("persists a batch once and reports the counts", async () => {
+      const persisted: EventRow[] = [];
+      const queue = new WriterPoolEventQueue({
+        persist: (event) => {
+          persisted.push(event);
+          return true;
+        },
+      });
+
+      const result = await queue.submit(rows(1, 3));
+
+      expect(result).toEqual({
+        queuedCount: 3,
+        insertedCount: 3,
+        duplicateCount: 0,
+      });
+      expect(persisted).toHaveLength(3);
+      expect(queue.size).toBe(0);
+      expect(queue.heldLockCount).toBe(0);
+      expect(queue.persistedKeyCount).toBe(3);
+    });
+
+    it("collapses duplicates inside a single batch", async () => {
+      let persistCalls = 0;
+      const queue = new WriterPoolEventQueue({
+        persist: () => {
+          persistCalls++;
+          return true;
+        },
+      });
+
+      const result = await queue.submit([row(5), row(5), row(5)]);
+
+      expect(persistCalls).toBe(1);
+      expect(result.queuedCount).toBe(1);
+      expect(result.insertedCount).toBe(1);
+      expect(result.duplicateCount).toBe(2);
+    });
+
+    it("persists each event exactly once under concurrent submits", async () => {
+      const persistCounts = new Map<string, number>();
+      const queue = new WriterPoolEventQueue({
+        persist: async (event) => {
+          const key = writerPoolEventIdentityKey(event);
+          // Yield inside the critical section: without a lock this is exactly
+          // where a second caller would slip in and insert the same row.
+          await sleep(2);
+          persistCounts.set(key, (persistCounts.get(key) ?? 0) + 1);
+          return true;
+        },
+      });
+
+      const batch = rows(1, 5);
+      const results = await Promise.all(
+        Array.from({ length: 8 }, () => queue.submit(batch)),
+      );
+
+      expect(persistCounts.size).toBe(5);
+      expect([...persistCounts.values()].every((n) => n === 1)).toBe(true);
+
+      const inserted = results.reduce((sum, r) => sum + r.insertedCount, 0);
+      const duplicates = results.reduce((sum, r) => sum + r.duplicateCount, 0);
+      expect(inserted).toBe(5);
+      expect(inserted + duplicates).toBe(8 * 5);
+      expect(queue.size).toBe(0);
+      expect(queue.heldLockCount).toBe(0);
+    });
+
+    it("never runs two persists for the same event at the same time", async () => {
+      const inFlight = new Set<string>();
+      let overlaps = 0;
+      const queue = new WriterPoolEventQueue({
+        persist: async (event) => {
+          const key = writerPoolEventIdentityKey(event);
+          if (inFlight.has(key)) overlaps++;
+          inFlight.add(key);
+          await sleep(2);
+          inFlight.delete(key);
+          return true;
+        },
+      });
+
+      await Promise.all([
+        queue.submit(rows(1, 4)),
+        queue.submit(rows(1, 4)),
+        queue.submit(rows(1, 4)),
+      ]);
+
+      expect(overlaps).toBe(0);
+    });
+
+    it("lets unrelated events persist concurrently", async () => {
+      let active = 0;
+      let peak = 0;
+      const queue = new WriterPoolEventQueue({
+        persist: async () => {
+          active++;
+          peak = Math.max(peak, active);
+          await sleep(5);
+          active--;
+          return true;
+        },
+      });
+
+      await Promise.all([
+        queue.submit([row(1, "initialized")]),
+        queue.submit([row(2, "initialized")]),
+        queue.submit([row(3, "initialized")]),
+      ]);
+
+      expect(peak).toBeGreaterThan(1);
+    });
+
+    it("counts a persist that reports no write as a duplicate", async () => {
+      const queue = new WriterPoolEventQueue({ persist: () => false });
+
+      const result = await queue.submit([row(9)]);
+
+      expect(result.insertedCount).toBe(0);
+      expect(result.duplicateCount).toBe(1);
+    });
+
+    it("releases the lock when a persist throws", async () => {
+      let calls = 0;
+      const queue = new WriterPoolEventQueue({
+        persist: () => {
+          calls++;
+          if (calls === 1) throw new Error("database is locked");
+          return true;
+        },
+      });
+
+      await expect(queue.submit([row(4)])).rejects.toThrow("database is locked");
+      expect(queue.heldLockCount).toBe(0);
+
+      const retry = await queue.submit([row(4)]);
+      expect(retry.insertedCount).toBe(1);
+      expect(queue.heldLockCount).toBe(0);
+    });
+
+    it("releases the lock after a successful persist", async () => {
+      const queue = new WriterPoolEventQueue({ persist: () => true });
+
+      await queue.submit([row(1), row(2)]);
+
+      expect(queue.heldLockCount).toBe(0);
+      expect(queue.size).toBe(0);
+    });
+
+    it("queues notifications before a flush and drains them once", async () => {
+      const persisted: EventRow[] = [];
+      const queue = new WriterPoolEventQueue({
+        persist: (event) => {
+          persisted.push(event);
+          return true;
+        },
+      });
+
+      const first = await queue.enqueue(rows(1, 3));
+      const second = await queue.enqueue(rows(2, 4));
+
+      // Ledgers 2 and 3 are already queued, so only ledger 4 is new.
+      expect(first.queuedCount).toBe(3);
+      expect(second.queuedCount).toBe(1);
+      expect(second.duplicateCount).toBe(2);
+      expect(queue.size).toBe(4);
+
+      const flushed = await queue.flush();
+      expect(flushed.processedCount).toBe(4);
+      expect(flushed.insertedCount).toBe(4);
+      expect(persisted).toHaveLength(4);
+      expect(queue.size).toBe(0);
+    });
+
+    it("does not re-persist an event enqueued again after a flush", async () => {
+      let persistCalls = 0;
+      const queue = new WriterPoolEventQueue({
+        persist: () => {
+          persistCalls++;
+          return true;
+        },
+      });
+
+      await queue.submit([row(11)]);
+      const again = await queue.enqueue([row(11)]);
+
+      expect(again.queuedCount).toBe(0);
+      expect(again.duplicateCount).toBe(1);
+      expect(persistCalls).toBe(1);
+      expect(queue.hasPersisted(row(11))).toBe(true);
+    });
+
+    it("drains safely when flushes run concurrently", async () => {
+      let persistCalls = 0;
+      const queue = new WriterPoolEventQueue({
+        persist: async () => {
+          persistCalls++;
+          await sleep(1);
+          return true;
+        },
+      });
+
+      await queue.enqueue(rows(1, 20));
+      await Promise.all([queue.flush(), queue.flush(), queue.flush()]);
+
+      expect(persistCalls).toBe(20);
+      expect(queue.size).toBe(0);
+      expect(queue.persistedKeyCount).toBe(20);
+    });
+
+    it("rejects an enqueue past maxQueueSize", async () => {
+      const queue = new WriterPoolEventQueue({
+        persist: () => true,
+        maxQueueSize: 3,
+      });
+
+      await expect(queue.enqueue(rows(1, 5))).rejects.toThrow(
+        WriterPoolEventQueueOverflowError,
+      );
+      expect(queue.size).toBe(3);
+    });
+
+    it("rejects an invalid maxQueueSize", () => {
+      expect(() => new WriterPoolEventQueue({ maxQueueSize: 0 })).toThrow(
+        /maxQueueSize must be a positive integer/,
+      );
+    });
+
+    it("clears all state on reset", async () => {
+      const queue = new WriterPoolEventQueue({ persist: () => true });
+      await queue.submit(rows(1, 3));
+
+      queue.reset();
+
+      expect(queue.size).toBe(0);
+      expect(queue.persistedKeyCount).toBe(0);
+      expect(queue.hasPersisted(row(1, "initialized"))).toBe(false);
+      expect(queue.heldLockCount).toBe(0);
+    });
+
+    it("exposes the default queue ceiling", () => {
+      expect(DEFAULT_WRITER_POOL_EVENT_QUEUE_MAX_SIZE).toBe(10_000);
+    });
+
+    it("survives repeated waves of concurrent identical batches", async () => {
+      const persistCounts = new Map<string, number>();
+      const queue = new WriterPoolEventQueue({
+        persist: async (event) => {
+          const key = writerPoolEventIdentityKey(event);
+          await sleep(1);
+          persistCounts.set(key, (persistCounts.get(key) ?? 0) + 1);
+          return true;
+        },
+      });
+
+      const batch = rows(1, 6, 2);
+      for (let wave = 0; wave < 4; wave++) {
+        await Promise.all(Array.from({ length: 5 }, () => queue.submit(batch)));
+      }
+
+      expect(persistCounts.size).toBe(12);
+      expect([...persistCounts.values()].every((n) => n === 1)).toBe(true);
+      expect(queue.heldLockCount).toBe(0);
+    });
+  });
+
+  describe("queueWrite lock release", () => {
+    it("releases the write-queue lock after a failed operation so later writes proceed", async () => {
+      const failOp: WriteOperation<void> = {
+        name: "failing-lock-release",
+        execute: () => {
+          throw new Error("intentional writer failure");
+        },
+      };
+
+      const failed = await queueWrite(failOp);
+      expect(failed.success).toBe(false);
+
+      const recovered = await queueWrite({
+        name: "after-failure",
+        execute: () => 1,
+      });
+      expect(recovered.success).toBe(true);
+      expect(recovered.data).toBe(1);
+    });
+
+    it("does not serialize unrelated event identities through the event lock", async () => {
+      let active = 0;
+      let peak = 0;
+      const queue = new WriterPoolEventQueue({
+        persist: async () => {
+          active++;
+          peak = Math.max(peak, active);
+          await sleep(8);
+          active--;
+          return true;
+        },
+      });
+
+      await Promise.all(
+        Array.from({ length: 6 }, (_, i) =>
+          queue.submit([row(100 + i, "funded")]),
+        ),
+      );
+
+      expect(peak).toBeGreaterThan(1);
+      expect(queue.heldLockCount).toBe(0);
+    });
+  });
+
+  describe("with a real SQLite store", () => {
+    let testDb: Database.Database;
+
+    beforeEach(async () => {
+      testDb = new Database(":memory:");
+      setDb(testDb);
+      runMigrations();
+      resetWriterPoolStartState();
+      await flushWriteQueue();
+    });
+
+    afterEach(async () => {
+      await flushWriteQueue();
+      resetWriterPoolStartState();
+      closeDb();
+    });
+
+    function eventCount(): number {
+      return (testDb.prepare("SELECT COUNT(*) AS c FROM events").get() as any).c;
+    }
+
+    it("does not duplicate entries when the same notifications arrive concurrently", async () => {
+      const batch = rows(1, 10, 2);
+
+      const results = await Promise.all(
+        Array.from({ length: 6 }, () => submitEventNotifications(batch)),
+      );
+
+      expect(eventCount()).toBe(20);
+      expect(results.reduce((sum, r) => sum + r.insertedCount, 0)).toBe(20);
+
+      const perLedger = testDb
+        .prepare(
+          `SELECT ledger_sequence AS ledger, COUNT(*) AS c FROM events
+           GROUP BY ledger_sequence ORDER BY ledger_sequence`,
+        )
+        .all() as Array<{ ledger: number; c: number }>;
+      expect(perLedger).toHaveLength(10);
+      expect(perLedger.every((r) => r.c === 2)).toBe(true);
+    });
+
+    it("keeps concurrent event notifications free of duplicate rows", async () => {
+      const batch = rows(1, 8, 3);
+
+      await Promise.all([
+        submitEventNotifications(batch),
+        submitEventNotifications(batch),
+        submitEventNotifications(batch),
+        submitEventNotifications(batch),
+      ]);
+
+      expect(eventCount()).toBe(24);
+      expect(getWriterPoolEventQueue().heldLockCount).toBe(0);
+    });
+
+    it("is idempotent across sequential submissions of the same window", async () => {
+      const batch = rows(100, 104, 3);
+
+      const first = await submitEventNotifications(batch);
+      const second = await submitEventNotifications(batch);
+
+      expect(first.insertedCount).toBe(15);
+      expect(second.insertedCount).toBe(0);
+      expect(second.duplicateCount).toBe(15);
+      expect(eventCount()).toBe(15);
+    });
+
+    it("still de-duplicates after a state reset, because the store rejects the row", async () => {
+      const batch = rows(200, 202, 2);
+
+      await submitEventNotifications(batch);
+      resetWriterPoolStartState();
+      const second = await submitEventNotifications(batch);
+
+      expect(second.insertedCount).toBe(0);
+      expect(second.duplicateCount).toBe(6);
+      expect(eventCount()).toBe(6);
+    });
+
+    it("does not double-count rows written outside the queue", async () => {
+      insertEvent(CONTRACT_ID, "funded", 1, 1_700_000_001, "{}");
+
+      const result = await submitEventNotifications([row(1, "funded")]);
+
+      expect(result.insertedCount).toBe(0);
+      expect(result.duplicateCount).toBe(1);
+      expect(eventCount()).toBe(1);
+    });
+
+    it("does not duplicate when notifications race queueWrite inserts", async () => {
+      const batch = rows(50, 54, 2);
+
+      await Promise.all([
+        submitEventNotifications(batch),
+        submitEventNotifications(batch),
+        queueWrite({
+          name: "direct-insert-event",
+          execute: () =>
+            insertEvent(
+              CONTRACT_ID,
+              "funded",
+              52,
+              1_700_000_052,
+              JSON.stringify({ ledger: 52, eventType: "funded" }),
+            ),
+        }),
+      ]);
+
+      expect(eventCount()).toBe(10);
+    });
+
+    it("survives repeated concurrent submissions of the same window", async () => {
+      const batch = rows(300, 304, 2);
+
+      for (let wave = 0; wave < 3; wave++) {
+        await Promise.all(
+          Array.from({ length: 8 }, () => submitEventNotifications(batch)),
+        );
+      }
+
+      expect(eventCount()).toBe(10);
+      expect(getWriterPoolEventQueue().heldLockCount).toBe(0);
+      expect(getWriterPoolEventQueue().size).toBe(0);
+    });
+
+    it("propagates persist failures and still allows a later retry", async () => {
+      const queue = new WriterPoolEventQueue({
+        persist: async (event) => {
+          const result = await queueWrite({
+            name: "insert-event-may-fail",
+            execute: () => {
+              if (event.ledgerSequence === 1 && event.eventType === "boom") {
+                throw new Error("constraint boom");
+              }
+              return insertEvent(
+                event.contractId,
+                event.eventType,
+                event.ledgerSequence,
+                event.timestamp,
+                event.dataJson,
+              );
+            },
+          });
+          if (!result.success) {
+            throw result.error ?? new Error("insert failed");
+          }
+          return Boolean(result.data);
+        },
+      });
+
+      await expect(
+        queue.submit([
+          {
+            ...row(1, "boom"),
+            eventType: "boom",
+          },
+        ]),
+      ).rejects.toThrow("constraint boom");
+      expect(queue.heldLockCount).toBe(0);
+
+      const retry = await queue.submit([row(2, "funded")]);
+      expect(retry.insertedCount).toBe(1);
+      expect(eventCount()).toBe(1);
+    });
+  });
+});
diff --git a/src/indexer/database-writer-pool.ts b/src/indexer/database-writer-pool.ts
index b3c1046..9e00aa7 100644
--- a/src/indexer/database-writer-pool.ts
+++ b/src/indexer/database-writer-pool.ts
@@ -1,8 +1,10 @@
 import {
   getDb,
+  insertEvent,
   getShippedMigrationVersions,
   verifySchemaIntegrity,
   verifySchemaUpToDate,
+  type EventRow,
 } from "./db.js";
 import logger from "../utils/logger.js";
 
@@ -16,6 +18,7 @@ import logger from "../utils/logger.js";
  * - Automatic rollback on failures
  * - Queue-based serialization to prevent writer contention
  * - Built-in retry logic for transient conflicts
+ * - In-memory queue locks for concurrent event notifications (#327)
  * - Migration verification hooks that validate the schema before starting (#331)
  * - High-frequency debug diagnostics for write speeds and payload sizes (#328)
  */
@@ -406,6 +409,305 @@ export function createReadWriteOperation<T>(
   };
 }
 
+// ---------------------------------------------------------------------------
+// In-memory event queue locks (#327)
+// ---------------------------------------------------------------------------
+
+/** Default ceiling on event rows held in memory before an overflow is raised. */
+export const DEFAULT_WRITER_POOL_EVENT_QUEUE_MAX_SIZE = 10_000;
+
+export class WriterPoolEventQueueOverflowError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "WriterPoolEventQueueOverflowError";
+  }
+}
+
+/**
+ * Persist a single event row. Returns true when a new row was written and
+ * false when the store already held it. Defaults to a `queueWrite` +
+ * `INSERT OR IGNORE` so inserts still go through the writer pool.
+ */
+export type WriterPoolEventPersistFn = (
+  event: EventRow,
+) => boolean | Promise<boolean>;
+
+export interface WriterPoolEventQueueOptions {
+  persist?: WriterPoolEventPersistFn;
+  maxQueueSize?: number;
+  /** Instance name used in queue diagnostics. */
+  name?: string;
+}
+
+export interface WriterPoolEventEnqueueResult {
+  queuedCount: number;
+  duplicateCount: number;
+}
+
+export interface WriterPoolEventFlushResult {
+  processedCount: number;
+  insertedCount: number;
+  duplicateCount: number;
+}
+
+export interface WriterPoolEventSubmitResult {
+  queuedCount: number;
+  insertedCount: number;
+  duplicateCount: number;
+}
+
+/** Identity used by the events table UNIQUE(contract_id, ledger_sequence, event_type). */
+export function writerPoolEventIdentityKey(
+  event: Pick<EventRow, "contractId" | "eventType" | "ledgerSequence">,
+): string {
+  return `${event.contractId}|${event.ledgerSequence}|${event.eventType}`;
+}
+
+function validatePositiveInt(name: string, value: unknown): number {
+  if (typeof value !== "number" || !Number.isInteger(value) || value < 1) {
+    throw new Error(`${name} must be a positive integer, received ${String(value)}`);
+  }
+  return value;
+}
+
+/**
+ * Persist through the writer pool so concurrent notifications share the same
+ * transaction + retry path as every other `queueWrite` caller.
+ */
+async function defaultPersistEvent(event: EventRow): Promise<boolean> {
+  const result = await queueWrite({
+    name: "insert-event",
+    execute: () =>
+      insertEvent(
+        event.contractId,
+        event.eventType,
+        event.ledgerSequence,
+        event.timestamp,
+        event.dataJson,
+      ),
+  });
+  if (!result.success) {
+    throw result.error ?? new Error("insert-event failed");
+  }
+  return Boolean(result.data);
+}
+
+/**
+ * Bounded in-memory queue that serializes event inserts per event identity.
+ *
+ * Concurrent `database_writer_pool` notifications routinely carry the same
+ * event (overlapping poll windows, retried pages, several writers in one
+ * process). Without a lock, two callers can both observe "not indexed yet"
+ * and both insert. The queue closes that window: every row is drained under
+ * a lock keyed on `contractId|ledgerSequence|eventType`, and the persisted-
+ * key set is checked inside that lock, so exactly one caller writes each
+ * event. Unrelated events still persist concurrently.
+ */
+export class WriterPoolEventQueue {
+  readonly name: string;
+  readonly maxQueueSize: number;
+
+  private readonly persist: WriterPoolEventPersistFn;
+  private readonly pending: EventRow[] = [];
+  private readonly pendingKeys = new Set<string>();
+  private readonly persistedKeys = new Set<string>();
+  private readonly lockTails = new Map<string, Promise<void>>();
+  private readonly heldLocks = new Set<string>();
+  private queueMutex: Promise<void> = Promise.resolve();
+
+  constructor(options: WriterPoolEventQueueOptions = {}) {
+    this.name = options.name ?? "database_writer_pool";
+    this.persist = options.persist ?? defaultPersistEvent;
+    this.maxQueueSize = validatePositiveInt(
+      "maxQueueSize",
+      options.maxQueueSize ?? DEFAULT_WRITER_POOL_EVENT_QUEUE_MAX_SIZE,
+    );
+  }
+
+  /** Rows currently waiting to be flushed. */
+  get size(): number {
+    return this.pending.length;
+  }
+
+  /** Event locks held right now – exposed for concurrency assertions. */
+  get heldLockCount(): number {
+    return this.heldLocks.size;
+  }
+
+  /** Distinct event identities this queue has already persisted. */
+  get persistedKeyCount(): number {
+    return this.persistedKeys.size;
+  }
+
+  hasPersisted(
+    event: Pick<EventRow, "contractId" | "eventType" | "ledgerSequence">,
+  ): boolean {
+    return this.persistedKeys.has(writerPoolEventIdentityKey(event));
+  }
+
+  /** Drop all queue state. Intended for tests and process restarts. */
+  reset(): void {
+    this.pending.length = 0;
+    this.pendingKeys.clear();
+    this.persistedKeys.clear();
+    this.lockTails.clear();
+    this.heldLocks.clear();
+    this.queueMutex = Promise.resolve();
+  }
+
+  /**
+   * Serialize mutations of the queue structure itself, so concurrent
+   * enqueue/flush callers never interleave a read and a write of `pending`.
+   */
+  private async withQueueMutex<T>(fn: () => T): Promise<T> {
+    const previous = this.queueMutex;
+    let release!: () => void;
+    this.queueMutex = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    try {
+      await previous;
+      return fn();
+    } finally {
+      release();
+    }
+  }
+
+  /**
+   * Serialize work for one event identity. Unrelated keys run concurrently and
+   * the lock is always released, including when `fn` throws.
+   */
+  private async withEventLock<T>(
+    key: string,
+    fn: () => Promise<T> | T,
+  ): Promise<T> {
+    const previous = this.lockTails.get(key) ?? Promise.resolve();
+    let release!: () => void;
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const tail = previous.then(
+      () => gate,
+      () => gate,
+    );
+    this.lockTails.set(key, tail);
+
+    try {
+      await previous.catch(() => undefined);
+      this.heldLocks.add(key);
+      return await fn();
+    } finally {
+      this.heldLocks.delete(key);
+      release();
+      if (this.lockTails.get(key) === tail) {
+        this.lockTails.delete(key);
+      }
+    }
+  }
+
+  /**
+   * Queue rows for insertion, dropping any already queued or already
+   * persisted. Throws `WriterPoolEventQueueOverflowError` past `maxQueueSize`.
+   */
+  async enqueue(events: EventRow[]): Promise<WriterPoolEventEnqueueResult> {
+    return this.withQueueMutex(() => {
+      let queuedCount = 0;
+      let duplicateCount = 0;
+
+      for (const event of events) {
+        const key = writerPoolEventIdentityKey(event);
+        if (this.pendingKeys.has(key) || this.persistedKeys.has(key)) {
+          duplicateCount++;
+          continue;
+        }
+        if (this.pending.length >= this.maxQueueSize) {
+          throw new WriterPoolEventQueueOverflowError(
+            `${this.name} event queue is full (maxQueueSize=${this.maxQueueSize})`,
+          );
+        }
+        this.pendingKeys.add(key);
+        this.pending.push(event);
+        queuedCount++;
+      }
+
+      return { queuedCount, duplicateCount };
+    });
+  }
+
+  /** Drain the queue, persisting each row under its own event lock. */
+  async flush(): Promise<WriterPoolEventFlushResult> {
+    let processedCount = 0;
+    let insertedCount = 0;
+    let duplicateCount = 0;
+
+    for (;;) {
+      const next = await this.withQueueMutex(() => this.pending.shift());
+      if (!next) break;
+
+      const key = writerPoolEventIdentityKey(next);
+      processedCount++;
+
+      await this.withEventLock(key, async () => {
+        try {
+          if (this.persistedKeys.has(key)) {
+            duplicateCount++;
+            return;
+          }
+          const inserted = await this.persist(next);
+          this.persistedKeys.add(key);
+          if (inserted) {
+            insertedCount++;
+          } else {
+            duplicateCount++;
+          }
+        } finally {
+          this.pendingKeys.delete(key);
+        }
+      });
+    }
+
+    return { processedCount, insertedCount, duplicateCount };
+  }
+
+  /** Enqueue and flush in one step – the entry point for event notifications. */
+  async submit(events: EventRow[]): Promise<WriterPoolEventSubmitResult> {
+    const enqueued = await this.enqueue(events);
+    const flushed = await this.flush();
+
+    const result: WriterPoolEventSubmitResult = {
+      queuedCount: enqueued.queuedCount,
+      insertedCount: flushed.insertedCount,
+      duplicateCount: enqueued.duplicateCount + flushed.duplicateCount,
+    };
+
+    logger.debug("database_writer_pool event queue submit", {
+      queue: this.name,
+      submitted: events.length,
+      ...result,
+    });
+
+    return result;
+  }
+}
+
+const defaultEventQueue = new WriterPoolEventQueue();
+
+/** The process-wide queue used by `submitEventNotifications`. */
+export function getWriterPoolEventQueue(): WriterPoolEventQueue {
+  return defaultEventQueue;
+}
+
+/**
+ * Index a batch of event notifications through the locked memory queue and
+ * the writer pool. Concurrent callers sharing an event identity collapse to
+ * a single insert; unrelated identities proceed in parallel.
+ */
+export function submitEventNotifications(
+  events: EventRow[],
+): Promise<WriterPoolEventSubmitResult> {
+  return defaultEventQueue.submit(events);
+}
+
 // ---------------------------------------------------------------------------
 // Polling diagnostics (#328)
 // ---------------------------------------------------------------------------
@@ -701,10 +1003,11 @@ export function getWriterPoolSchemaReport(): WriterPoolSchemaReport | null {
   return lastSchemaReport;
 }
 
-/** Reset start/enforcement state and registered hooks. Intended for tests. */
+/** Reset start/enforcement state, registered hooks, and event-queue locks. Intended for tests. */
 export function resetWriterPoolStartState(): void {
   poolStarted = false;
   enforceStart = false;
   lastSchemaReport = null;
   migrationHooks.clear();
+  defaultEventQueue.reset();
 }