fix: Preserve transmit order across runtimes

Persist effect positions within each source message and preserve legacy rows during schema upgrades. Reject null transmit arguments and refresh the parity contract for message reads, administration, and wake-up behavior.

Author: cardmagic

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 ## Unreleased
 
+- Preserve transmit staging order within a source message through a persisted
+  effect position, including retries. Schema migration 14 adds `effects.position`;
+  run `runtime.install()` before starting upgraded workers. Legacy rows retain
+  their existing ID tie-break because their original order cannot be recovered.
+- Reject explicit null transmit arguments, matching Ruby; omission still defaults
+  to an empty object. Expand the shared wire fixtures and adapter ordering tests.
+- Refresh the parity ledger for authorized message reads and retained background
+  results, bulk redrive/audit records, reminder cancellation, and automatic wake-up.
+
 - Patch the Cloudflare development tooling's Undici dependency to 7.29.1 to resolve the high-severity WebSocket and TLS advisories reported by CI.
 
 - Add portable observability envelopes, metric definitions, actor observers, and bounded authorization-aware diagnostics across SQL runtimes and the Durable Objects host. Isolate failing instrumentation and error loggers from actor work.

compatibility/transmit-envelopes.json

@@ -84,6 +84,16 @@
         "operation": "increment",
         "arguments": [1]
       }
+    },
+    {
+      "name": "null arguments",
+      "envelope": {
+        "effectId": "fixture-effect-0008",
+        "actorType": "transmit-counters",
+        "actorId": "fixture-counter",
+        "operation": "increment",
+        "arguments": null
+      }
     }
   ]
 }

docs/api.md

@@ -105,7 +105,7 @@ envelope. The envelope carries its name in `invalidations`. A component registry
 can then refresh a reauthorized endpoint, and the value stays private.
 
 `MessageReference` does not retain an invocation's authorization context.
-Supply `authorizationContext` to each `status()`, `result()`, and `wait()` call;
+Supply `authorizationContext` to each `status()`, `result()`, `outcome()`, and `wait()` call;
 the runtime reauthorizes the persisted operation every time. Durable results
 are JSON, so an operation that returns `undefined` or is declared `void`
 resolves as `null`.
@@ -904,6 +904,11 @@ open SAH pool otherwise blocks the next candidate until the worker dies.
 
 ## `solid-objects/transmit`
 
+Schema migration 14 persists an effect's staging position within its source message.
+Run `runtime.install()` before upgraded workers start. New effects preserve staging
+order even when two transmits originate in one turn. Legacy rows use position zero
+and their existing ID tie-break; their original order cannot be recovered.
+
 The transactional outbox bridge between a local runtime and a server
 runtime. An actor stages a transmit intent with `this.transmit()`
 in the same transaction as its state change. The effect worker drains the
@@ -926,7 +931,7 @@ outbox with at-least-once delivery, per-actor order, and retry backoff.
   `runtime.deadLetters.retry` re-queues it.
 - `receiveTransmitEnvelope(options)`: idempotent server ingest. `arguments`
   is optional in the envelope and defaults to an empty object, matching the
-  staging side and the Ruby ingest. It enqueues an
+  staging side and the Ruby ingest. Explicit `null` and arrays are invalid. It enqueues an
   internal message with `transmit:<effectId>` as the idempotency key, so a
   replayed envelope applies once. The host must authenticate the sender
   before this call; internal delivery skips `authorizeMessage`. The call

docs/observability.md

@@ -43,6 +43,10 @@ Events describe local observations. Concurrent deletion, crashes, and failed
 exporters can omit events. Never infer exactly-once delivery from event counts.
 Additional existing runtime events retain their names.
 
+Portable `polling.interval_changed` events carry `previousIntervalMilliseconds`,
+`currentIntervalMilliseconds`, and a string `reason`. Ruby converts its native
+second-based notification values while preserving the original notification.
+
 ## Metrics and tracing
 
 Metrics are sample descriptions. Exporting them is opt-in: the runtime does not

docs/parity.md

@@ -115,8 +115,8 @@
     "test": "vitest run",
     "test:browser": "pnpm run build && playwright test",
     "test:coverage": "vitest run --coverage",
-    "test:postgresql": "vitest run test/postgresql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts",
-    "test:mysql": "vitest run test/mysql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts",
+    "test:postgresql": "vitest run test/postgresql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts",
+    "test:mysql": "vitest run test/mysql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts",
     "test:package": "node scripts/release-artifact-smoke.mjs",
     "test:recovery": "pnpm run build && node examples/failure-recovery/demo.ts",
     "test:at-least-once": "pnpm run build && node examples/at-least-once/demo.ts",

package.json

@@ -78,7 +78,7 @@ const EXPECTED_COLUMNS: Readonly<Record<string, readonly string[]>> = {
     "claimed_at_ms",
   ],
   reminders: ["id", "instance_id", "operation", "message_operation", "run_at_ms", "status"],
-  effects: ["id", "message_id", "instance_id", "name", "status", "available_at_ms"],
+  effects: ["id", "message_id", "instance_id", "name", "status", "available_at_ms", "position"],
   effect_recoveries: [
     "effect_id",
     "instance_id",

src/doctor.ts

@@ -731,13 +731,13 @@ export class Repository {
         turn.message.id,
       ])
 
-      for (const effect of input.intents.effects) {
+      for (const [position, effect] of input.intents.effects.entries()) {
         const effectId = effect.id ?? randomUUID()
         await connection.run(
           `INSERT INTO ${this.table("effects")}
            (id, message_id, instance_id, name, arguments, success_operation, failure_operation,
-            status, max_attempts, available_at_ms)
-           VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)`,
+            status, max_attempts, available_at_ms, position)
+           VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?)`,
           [
             effectId,
             turn.message.id,
@@ -748,6 +748,7 @@ export class Repository {
             effect.failureOperation ?? null,
             this.settings.maxAttempts,
             now,
+            position,
           ],
         )
         if (effect.recoveryOperation !== undefined || effect.statusOperation !== undefined) {

src/repository.ts

@@ -14,7 +14,8 @@ const INSTANCE_RETENTION_INDEX_VERSION = 10
 const DEAD_LETTER_REDRIVE_VERSION = 11
 const REQUEST_ID_LOOKUP_VERSION = 12
 const REMEMBERED_KEYS_VERSION = 13
-const LATEST_VERSION = REMEMBERED_KEYS_VERSION
+const EFFECT_POSITION_VERSION = 14
+const LATEST_VERSION = EFFECT_POSITION_VERSION
 
 export const SCHEMA_VERSIONS: readonly number[] = Object.freeze(
   Array.from({ length: LATEST_VERSION }, (_unused, index) => index + 1),
@@ -390,6 +391,16 @@ export async function installSchema(options: {
     })
   }
 
+  if (!installedVersions.has(EFFECT_POSITION_VERSION)) {
+    await addEffectPosition({ connection, family, table: table("effects") })
+    await recordMigration({
+      connection,
+      table: table("schema_migrations"),
+      version: EFFECT_POSITION_VERSION,
+      schemaIdentity,
+    })
+  }
+
   if (installedVersions.has(POLLING_INDEXES_VERSION)) return
   const pollingIndexes = [
     ["effects", `${prefix}effects_poll`, "status, available_at_ms, id"],
@@ -471,12 +482,23 @@ async function installRedrive(options: {
   })
 }
 
+async function addEffectPosition(options: {
+  connection: DatabaseConnection
+  family: DatabaseFamily
+  table: string
+}): Promise<void> {
+  if (await hasColumn({ ...options, column: "position" })) return
+  await options.connection.run(
+    `ALTER TABLE ${options.table} ADD COLUMN position INTEGER NOT NULL DEFAULT 0`,
+  )
+}
+
 async function addFailedAt(options: {
   connection: DatabaseConnection
   family: DatabaseFamily
   table: string
 }): Promise<void> {
-  if (await hasFailedAt(options)) return
+  if (await hasColumn({ ...options, column: "failed_at_ms" })) return
 
   const type = options.family === "sqlite" ? "INTEGER" : "BIGINT"
   await options.connection.run(`ALTER TABLE ${options.table} ADD COLUMN failed_at_ms ${type}`)
@@ -485,22 +507,23 @@ async function addFailedAt(options: {
   )
 }
 
-async function hasFailedAt(options: {
+async function hasColumn(options: {
   connection: DatabaseConnection
   family: DatabaseFamily
   table: string
+  column: string
 }): Promise<boolean> {
   if (options.family === "sqlite") {
     const columns = await options.connection.all<{ name: string }>(
       `PRAGMA table_info(${options.table})`,
     )
-    return columns.some(({ name }) => name === "failed_at_ms")
+    return columns.some(({ name }) => name === options.column)
   }
   const schema = options.family === "postgresql" ? "current_schema()" : "DATABASE()"
   const found = await options.connection.get<{ found: number | bigint }>(
     `SELECT COUNT(*) AS found FROM information_schema.columns
-     WHERE table_schema = ${schema} AND table_name = ? AND column_name = 'failed_at_ms'`,
-    [options.table],
+     WHERE table_schema = ${schema} AND table_name = ? AND column_name = ?`,
+    [options.table, options.column],
   )
   return Number(found?.found ?? 0) > 0
 }

src/schema.ts

@@ -50,7 +50,7 @@ export async function receiveTransmitEnvelope(options: {
       throw new InvalidPayload(`transmit envelope requires a non-empty ${field}`)
     }
   }
-  const argumentsValue = envelope.arguments ?? {}
+  const argumentsValue = envelope.arguments === undefined ? {} : envelope.arguments
   if (!isJsonObject(argumentsValue)) {
     throw new InvalidPayload("transmit envelope arguments must be a JSON object")
   }
@@ -115,7 +115,7 @@ async function undeliveredEnvelopesThrough(input: {
          AND messages.actor_type = ?
          AND messages.actor_id = ?
          AND messages.sequence <= (SELECT sequence FROM ${messages} WHERE id = ?)
-       ORDER BY messages.sequence, effects.id`,
+       ORDER BY messages.sequence, effects.position, effects.id`,
       [effectName, context.actorType, context.actorId, context.sourceMessageId],
     ),
   )