Separate request identity

Author: cardmagic

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- Separate generated request identity from caller-supplied idempotency keys and
+  expose both with enqueue time through immutable actor message context.
 - Preserve isolated runtime ownership across operation, lifecycle, observable,
   and payload callbacks while keeping lifecycle callbacks message-free.
 - Retry transient SQLite writer acquisition with bounded capped backoff without

README.md

@@ -265,6 +265,10 @@ await counter
 
 Invocation options stay separate from actor arguments, so an actor may safely
 use argument names such as `timeoutMilliseconds` or `authorizationContext`.
+Every invocation receives a generated `requestId`; `idempotencyKey` remains the
+caller's deduplication key and is never reused as request identity. During an
+operation, `this.currentMessage` exposes both values along with `id`,
+`enqueuedAt`, `actorType`, `actorId`, `sequence`, and `attempt`.
 
 Do not make a committed actor call or wait on a message from inside
 `database.transaction(...)` on the Solid Objects database. The runtime raises

docs/architecture.md

@@ -6,7 +6,10 @@ persists JSON state between activations.
 
 A durable message envelope selects an actor `operation` and records its
 `delivery_mode` as `async`, `sync`, or `internal`. An operation is actor code;
-a message is the durable delivery record that invokes it.
+a message is the durable delivery record that invokes it. Each message has a
+generated request ID independent of its optional caller-supplied idempotency
+key. Matching idempotency keys are scoped to one actor and must identify the
+same operation, delivery mode, and arguments.
 
 The correctness contract is:
 

docs/parity.md

@@ -23,7 +23,7 @@ Reference: Ruby `solid_objects` 0.12.0 at commit `a01b6f5`.
 | ------------------------------------------------------------------------------------------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
 | Actor registry, durable identity, JSON state, and adjacent state migrations                                   | Native  | Ordinary classes, static actor types, inferred state, explicit migrations, and isolated runtime context across every actor-instance callback.                                                                      |
 | Fluent committed calls and background delivery                                                                | Native  | `await reference.operation()` and `reference.send.operation()`.                                                                                                                                                    |
-| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native  | Relational ready/claimed membership tables, durable history, and adapter-appropriate sequence locking.                                                                                                             |
+| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native  | Relational ready/claimed membership tables, distinct generated request IDs and caller idempotency keys, durable history, and adapter-appropriate sequence locking.                                                 |
 | Domain rejection and strict poison ordering                                                                   | Native  | Rejections roll back without retry; retryable failures block later operations until completion or dead-lettering.                                                                                                  |
 | Bounded activation passes and hot-actor fairness                                                              | Native  | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting.                                                               |
 | Bounded claim candidate scan                                                                                  | Native  | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race.                                                                                                 |

src/doctor.ts

@@ -44,6 +44,7 @@ const EXPECTED_COLUMNS: Readonly<Record<string, readonly string[]>> = {
   messages: [
     "id",
     "request_id",
+    "idempotency_key",
     "instance_id",
     "operation",
     "delivery_mode",
@@ -182,11 +183,11 @@ export class Doctor {
           message: `incompatible schema identity ${wrongIdentity.schema_identity}`,
         })
       }
-      if (versions.join(",") !== "1,2") {
+      if (versions.join(",") !== "1,2,3") {
         return check({
           name: "schema",
           status: "fail",
-          message: `expected schema migrations 1, 2; found ${versions.join(", ")}`,
+          message: `expected schema migrations 1, 2, 3; found ${versions.join(", ")}`,
         })
       }
       return check({

src/records.ts

@@ -29,6 +29,7 @@ export interface InstanceRow {
 export interface MessageRow {
   id: string
   request_id: string
+  idempotency_key: string | null
   instance_id: string
   actor_type: string
   actor_id: string
@@ -42,6 +43,8 @@ export interface MessageRow {
   attempt_count: number | bigint
   max_attempts: number | bigint
   completed_at_ms: number | bigint | null
+  created_at_ms: number | bigint
+  updated_at_ms: number | bigint
 }
 
 export interface DeadLetterRow {

src/repository.ts

@@ -221,14 +221,15 @@ export class Repository {
     }
     if (!instance) throw new ActorDestroyed("actor disappeared during enqueue")
 
-    if (input.idempotencyKey) {
+    if (input.idempotencyKey !== undefined) {
       const existing = await connection.get<MessageRow>(
-        `SELECT * FROM ${this.table("messages")} WHERE actor_type = ? AND actor_id = ? AND request_id = ?`,
+        `SELECT * FROM ${this.table("messages")} WHERE actor_type = ? AND actor_id = ? AND idempotency_key = ?`,
         [input.actorType, input.actorId, input.idempotencyKey],
       )
       if (existing) {
         if (
           existing.operation !== input.operation ||
+          existing.delivery_mode !== input.deliveryMode ||
           existing.arguments !== JSON.stringify(input.arguments)
         ) {
           throw new IdempotencyConflict("idempotency key identifies a different invocation")
@@ -253,20 +254,21 @@ export class Repository {
 
     const sequence = BigInt(instance.next_message_sequence)
     const messageId = randomUUID()
-    const requestId = input.idempotencyKey ?? randomUUID()
+    const requestId = randomUUID()
     await connection.run(
       `UPDATE ${this.table("instances")} SET next_message_sequence = next_message_sequence + 1,
        updated_at_ms = ? WHERE id = ?`,
       [now, instance.id],
     )
     await connection.run(
       `INSERT INTO ${this.table("messages")}
-       (id, request_id, instance_id, actor_type, actor_id, sequence, operation, delivery_mode,
-        arguments, max_attempts, created_at_ms, updated_at_ms)
-       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+       (id, request_id, idempotency_key, instance_id, actor_type, actor_id, sequence, operation,
+        delivery_mode, arguments, max_attempts, created_at_ms, updated_at_ms)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
       [
         messageId,
         requestId,
+        input.idempotencyKey ?? null,
         instance.id,
         input.actorType,
         input.actorId,

src/runtime.ts

@@ -1783,14 +1783,16 @@ function messageInstrumentation(message: MessageRow): JsonObject {
 }
 
 function actorMessageContext(message: MessageRow): MessageContext {
-  return {
+  return Object.freeze({
     id: message.id,
     requestId: message.request_id,
+    idempotencyKey: message.idempotency_key,
+    enqueuedAt: new Date(Number(message.created_at_ms)),
     actorType: message.actor_type,
     actorId: message.actor_id,
     sequence: BigInt(message.sequence),
     attempt: Number(message.attempt_count),
-  }
+  })
 }
 
 function restoreActorState(options: {

src/schema.ts

@@ -2,7 +2,9 @@ import { UnsupportedDatabase } from "./errors.js"
 import type { DatabaseConnection, DatabaseFamily } from "./database/types.js"
 
 const BASE_VERSION = 1
-const LATEST_VERSION = 2
+const RETRY_LINK_VERSION = 2
+const MESSAGE_IDENTITY_VERSION = 3
+const LATEST_VERSION = MESSAGE_IDENTITY_VERSION
 
 export async function installSchema(options: {
   connection: DatabaseConnection
@@ -189,27 +191,61 @@ export async function installSchema(options: {
     [BASE_VERSION, schemaIdentity, now],
   )
 
-  const retryLinkMigration = installedMigrations.some(
-    ({ version }) => Number(version) === LATEST_VERSION,
-  )
-  if (retryLinkMigration) return
+  const installedVersions = new Set(installedMigrations.map(({ version }) => Number(version)))
+  if (!installedVersions.has(RETRY_LINK_VERSION)) {
+    await connection.run(
+      `ALTER TABLE ${table("dead_letters")} ADD COLUMN ${family === "postgresql" ? "IF NOT EXISTS " : ""}retried_message_id ${family === "mysql" ? "VARCHAR(255)" : "TEXT"}
+       REFERENCES ${table("messages")}(id) ON DELETE SET NULL`,
+    )
+    await createIndex({
+      connection,
+      family,
+      table: table("dead_letters"),
+      name: `${prefix}dead_letters_retried_message`,
+      columns: "retried_message_id",
+    })
+    await recordMigration({
+      connection,
+      table: table("schema_migrations"),
+      version: RETRY_LINK_VERSION,
+      schemaIdentity,
+    })
+  }
 
+  if (installedVersions.has(MESSAGE_IDENTITY_VERSION)) return
+  await connection.run(
+    `ALTER TABLE ${table("messages")} ADD COLUMN ${family === "postgresql" ? "IF NOT EXISTS " : ""}idempotency_key ${family === "mysql" ? "VARCHAR(255)" : "TEXT"}`,
+  )
   await connection.run(
-    `ALTER TABLE ${table("dead_letters")} ADD COLUMN ${family === "postgresql" ? "IF NOT EXISTS " : ""}retried_message_id ${family === "mysql" ? "VARCHAR(255)" : "TEXT"}
-     REFERENCES ${table("messages")}(id) ON DELETE SET NULL`,
+    `UPDATE ${table("messages")} SET idempotency_key = request_id WHERE idempotency_key IS NULL`,
   )
   await createIndex({
     connection,
     family,
-    table: table("dead_letters"),
-    name: `${prefix}dead_letters_retried_message`,
-    columns: "retried_message_id",
+    table: table("messages"),
+    name: `${prefix}messages_idempotency`,
+    columns: "actor_type, actor_id, idempotency_key",
+    kind: "unique",
   })
-  const migratedAt = await connection.nowMilliseconds()
-  await connection.run(
-    `INSERT INTO ${table("schema_migrations")}(version, schema_identity, installed_at_ms)
+  await recordMigration({
+    connection,
+    table: table("schema_migrations"),
+    version: MESSAGE_IDENTITY_VERSION,
+    schemaIdentity,
+  })
+}
+
+async function recordMigration(options: {
+  connection: DatabaseConnection
+  table: string
+  version: number
+  schemaIdentity: string
+}): Promise<void> {
+  const installedAt = await options.connection.nowMilliseconds()
+  await options.connection.run(
+    `INSERT INTO ${options.table}(version, schema_identity, installed_at_ms)
      VALUES (?, ?, ?) ON CONFLICT(version) DO NOTHING`,
-    [LATEST_VERSION, schemaIdentity, migratedAt],
+    [options.version, options.schemaIdentity, installedAt],
   )
 }
 
@@ -233,10 +269,12 @@ async function createIndex(options: {
   table: string
   name: string
   columns: string
+  kind?: "unique"
 }): Promise<void> {
+  const kind = options.kind === "unique" ? "UNIQUE " : ""
   if (options.family !== "mysql") {
     await options.connection.run(
-      `CREATE INDEX IF NOT EXISTS ${options.name} ON ${options.table}(${options.columns})`,
+      `CREATE ${kind}INDEX IF NOT EXISTS ${options.name} ON ${options.table}(${options.columns})`,
     )
     return
   }
@@ -247,6 +285,6 @@ async function createIndex(options: {
   )
   if (Number(existing?.present ?? 0) > 0) return
   await options.connection.run(
-    `CREATE INDEX ${options.name} ON ${options.table}(${options.columns})`,
+    `CREATE ${kind}INDEX ${options.name} ON ${options.table}(${options.columns})`,
   )
 }

src/types.ts

@@ -37,12 +37,14 @@ export interface AdministrationOptions<AuthorizationContext = unknown> {
 }
 
 export interface MessageContext {
-  id: string
-  requestId: string
-  actorType: string
-  actorId: string
-  sequence: bigint
-  attempt: number
+  readonly id: string
+  readonly requestId: string
+  readonly idempotencyKey: string | null
+  readonly enqueuedAt: Date
+  readonly actorType: string
+  readonly actorId: string
+  readonly sequence: bigint
+  readonly attempt: number
 }
 
 export interface EffectContext {