#76 Surface metadata-missing as a typed failure inside persisted entity handlers

packages/local-browser/src/ReplicaRpc.ts

@@ -24,7 +24,7 @@ const ExportedDocument = Schema.Struct({
 
 const JsonOutcome = CommandOutcome.schema(Schema.Json, Schema.Json)
 const DocumentIdOutcome = CommandOutcome.schema(Identity.DocumentId, Schema.Never)
-export const protocolVersion = 4
+export const protocolVersion = 5
 const SessionLease = Schema.Struct({ leaseMillis: Schema.Int })
 export const SessionHandshake = Schema.Struct({
   leaseMillis: Schema.Int,

packages/local-browser/src/internal/restoreProtocol.ts

@@ -58,6 +58,7 @@ const CanonicalEncodeError = Schema.TaggedStruct("CanonicalEncodeError", {
 const StorageCorrupt = Schema.TaggedStruct("StorageCorrupt", {
   cause: BoundedErrorDescription
 })
+const ReplicaMetadataMissing = Schema.TaggedStruct("ReplicaMetadataMissing", {})
 const QuotaExceeded = Schema.TaggedStruct("QuotaExceeded", {
   resource: Schema.String,
   limit: Schema.Int
@@ -116,6 +117,7 @@ export const RestoreWireError = Schema.Union([
   StorageUnavailable,
   CanonicalEncodeError,
   StorageCorrupt,
+  ReplicaMetadataMissing,
   QuotaExceeded,
   MigrationFailed,
   BackupInvalid,
@@ -216,6 +218,7 @@ export const restoreWireErrorFields = fieldMetadata({
   StorageUnavailable,
   CanonicalEncodeError,
   StorageCorrupt,
+  ReplicaMetadataMissing,
   QuotaExceeded,
   MigrationFailed,
   BackupInvalid,
@@ -427,6 +430,12 @@ export const encodeReplicaError = (
         { _tag: reason._tag, cause: emptyErrorDescription },
         ({ defect }) => ({ _tag: reason._tag, cause: defect(reason.cause) })
       )
+    case "ReplicaMetadataMissing":
+      return encodeWithinBudget(
+        maxBytes,
+        { _tag: reason._tag },
+        () => ({ _tag: reason._tag })
+      )
     case "QuotaExceeded":
       return encodeWithinBudget(
         maxBytes,
@@ -599,6 +608,9 @@ export const replicaErrorFromWire = (wire: RestoreWireError): ReplicaError.Repli
     case "StorageCorrupt":
       reason = new ReplicaError.StorageCorrupt({ cause: decodeDefect(wire.cause) })
       break
+    case "ReplicaMetadataMissing":
+      reason = new ReplicaError.ReplicaMetadataMissing()
+      break
     case "QuotaExceeded":
       reason = new ReplicaError.QuotaExceeded({ resource: wire.resource, limit: wire.limit })
       break

packages/local-browser/test/ReplicaClient.test.ts

@@ -100,6 +100,10 @@ it.layer(NodeCrypto.layer)("ReplicaClient", (it) => {
     typeof ReplicaRpc.group,
     RpcClientError.RpcClientError
   >
+  const resolveRpcResponse = <A, E, R,>(
+    effect: Effect.Effect<A | Deferred.Deferred<A, E>, E, R>
+  ) =>
+    Effect.flatMap(effect, (value) => Deferred.isDeferred<A, E>(value) ? Deferred.await(value) : Effect.succeed(value))
   type FinishRestore = TestReplicaRpcClient["FinishRestoreBackupV4"]
   const dropTerminalReady = (
     rpc: TestReplicaRpcClient,
@@ -236,6 +240,8 @@ it.layer(NodeCrypto.layer)("ReplicaClient", (it) => {
 
   it.effect("decodes and rejects owners using an older protocol", () =>
     Effect.scoped(Effect.gen(function*() {
+      const sessions = yield* SessionManager.SessionManager
+      assert.strictEqual(ReplicaRpc.protocolVersion, 5)
       const open = ReplicaRpc.group.requests.get("OpenSession")
       if (open?._tag !== "OpenSession") return yield* Effect.die(new Error("OpenSession RPC not found"))
       yield* Schema.decodeUnknownEffect(open.successSchema)({
@@ -263,8 +269,64 @@ it.layer(NodeCrypto.layer)("ReplicaClient", (it) => {
       })
       const error = yield* Effect.flip(ReplicaClient.fromRpcClient(definition, older))
       assert.strictEqual(error.reason._tag, "ProtocolMismatch")
+      if (error.reason._tag === "ProtocolMismatch") {
+        assert.strictEqual(error.reason.expected, "protocol version 5")
+        assert.strictEqual(error.reason.observed, "protocol version 4")
+      }
+      assert.strictEqual(yield* sessions.activeCount, 0)
+    })).pipe(Effect.provide(Owner)))
+
+  it.effect("decodes and rejects owners using a newer protocol", () =>
+    Effect.scoped(Effect.gen(function*() {
+      const sessions = yield* SessionManager.SessionManager
+      const rpc = yield* RpcTest.makeClient(ReplicaRpc.group)
+      const newer = new Proxy(rpc, {
+        get(target, property, receiver) {
+          const value = Reflect.get(target, property, receiver)
+          if (property !== "OpenSession") return value
+          return (payload: never) =>
+            value(payload).pipe(Effect.map((lease) => ({
+              ...(lease as {
+                readonly leaseMillis: number
+                readonly protocolVersion: number
+                readonly definitionHash: string
+                readonly ownerEpoch: string
+              }),
+              protocolVersion: ReplicaRpc.protocolVersion + 1
+            })))
+        }
+      })
+      const error = yield* Effect.flip(ReplicaClient.fromRpcClient(definition, newer))
+      assert.strictEqual(error.reason._tag, "ProtocolMismatch")
+      if (error.reason._tag === "ProtocolMismatch") {
+        assert.strictEqual(error.reason.expected, "protocol version 5")
+        assert.strictEqual(error.reason.observed, "protocol version 6")
+      }
+      assert.strictEqual(yield* sessions.activeCount, 0)
     })).pipe(Effect.provide(Owner)))
 
+  it.effect("owner rejects an explicit older client protocol before opening a session", () =>
+    Effect.gen(function*() {
+      const open = yield* ReplicaRpc.group.accessHandler("OpenSession")
+      const sessions = yield* SessionManager.SessionManager
+      const response = open({
+        sessionId: yield* Identity.makeSessionId,
+        protocolVersion: 4,
+        definitionHash: definition.hash
+      }, {
+        client: new Rpc.ServerClient(1),
+        requestId: RequestId("old-client"),
+        headers: Headers.empty
+      })
+      const error = yield* Effect.flip(resolveRpcResponse(response))
+      assert.strictEqual(error.reason._tag, "ProtocolMismatch")
+      if (error.reason._tag === "ProtocolMismatch") {
+        assert.strictEqual(error.reason.expected, `5:${definition.hash}`)
+        assert.strictEqual(error.reason.observed, `4:${definition.hash}`)
+      }
+      assert.strictEqual(yield* sessions.activeCount, 0)
+    }).pipe(Effect.provide(Owner)))
+
   it.effect("recovers ambiguous commands through typed receipt lookup", () =>
     Effect.scoped(Effect.gen(function*() {
       const rpc = yield* RpcTest.makeClient(ReplicaRpc.group)

packages/local-browser/test/ReplicaOwnerRestoreV4.test.ts

@@ -99,7 +99,7 @@ it.layer(NodeCrypto.layer)("ReplicaOwner restore version 4", (it) => {
         protocolVersion: ReplicaRpc.protocolVersion,
         definitionHash: definition.hash
       })
-      assert.strictEqual(handshake.protocolVersion, 4)
+      assert.strictEqual(handshake.protocolVersion, 5)
       assert.strictEqual(handshake.maxChunkBytes, limits.maxChunkBytes)
       assert.strictEqual(handshake.maxRestoreCoalesceMillis, limits.maxRestoreCoalesceMillis)
       assert.strictEqual(handshake.maxRestoreErrorBytes, limits.maxRestoreErrorBytes)

packages/local-browser/test/RestoreClientProtocol.test.ts

@@ -1665,13 +1665,13 @@ it.layer(NodeCrypto.layer)("RestoreClientProtocol", (it) => {
       }
     }))
 
-  it.effect("rejects a version 3 handshake before opening the source transport", () =>
+  it.effect("rejects a version 4 handshake before opening the source transport", () =>
     Effect.scoped(Effect.gen(function*() {
       const oldRpc = {
         OpenSession: () =>
           Effect.succeed({
             leaseMillis: 10_000,
-            protocolVersion: 3,
+            protocolVersion: 4,
             definitionHash: definition.hash,
             ownerEpoch: "owner"
           }),
@@ -1680,8 +1680,8 @@ it.layer(NodeCrypto.layer)("RestoreClientProtocol", (it) => {
       const error = yield* ReplicaClient.fromRpcClient(definition, oldRpc).pipe(Effect.flip)
       assert.strictEqual(error.reason._tag, "ProtocolMismatch")
       if (error.reason._tag === "ProtocolMismatch") {
-        assert.strictEqual(error.reason.expected, "protocol version 4")
-        assert.strictEqual(error.reason.observed, "protocol version 3")
+        assert.strictEqual(error.reason.expected, "protocol version 5")
+        assert.strictEqual(error.reason.observed, "protocol version 4")
       }
     })))
 })

packages/local-browser/test/RestoreProtocol.test.ts

@@ -23,6 +23,16 @@ it.effect("round trips a typed restore error through its wire schema", () =>
     assert.deepStrictEqual(RestoreProtocol.replicaErrorFromWire(decoded), original)
   }))
 
+it.effect("reconstructs ReplicaMetadataMissing from the wire", () =>
+  Effect.gen(function*() {
+    const original = new ReplicaError.ReplicaError({
+      reason: new ReplicaError.ReplicaMetadataMissing()
+    })
+    const encoded = RestoreProtocol.encodeReplicaError(original, 4_096)
+    const decoded = yield* Schema.decodeUnknownEffect(RestoreProtocol.RestoreWireError)(encoded)
+    assert.deepStrictEqual(RestoreProtocol.replicaErrorFromWire(decoded), original)
+  }))
+
 it.effect("round trips a superseded checkpoint reason through the restore wire", () =>
   Effect.gen(function*() {
     const original = new ReplicaError.ReplicaError({
@@ -237,6 +247,7 @@ it.effect("encodes every restore error and defect at the minimum configured budg
       new ReplicaError.StorageUnavailable({ cause }),
       new ReplicaError.CanonicalEncodeError({ cause }),
       new ReplicaError.StorageCorrupt({ cause }),
+      new ReplicaError.ReplicaMetadataMissing(),
       new ReplicaError.QuotaExceeded({ resource: "resource".repeat(32), limit: 1 }),
       new ReplicaError.MigrationFailed({ migration: "migration".repeat(32), cause }),
       new ReplicaError.BackupInvalid({ cause }),

packages/local-rpc/test/PeerRpcServer.test.ts

@@ -263,6 +263,7 @@ const makeFixture = (options: {
       admit: Effect.acquireRelease(Effect.succeed(permit), () => Effect.void),
       claim: (use) => use(permit),
       refresh: Effect.succeed(permit),
+      preflight: () => Effect.void,
       validate: () => Effect.void
     })
     const sync = PeerSync.PeerSync.of({

packages/local-sql/src/Compaction.ts

@@ -3,6 +3,7 @@ import * as Canonical from "@lucas-barake/effect-local/Canonical"
 import type * as Document from "@lucas-barake/effect-local/Document"
 import * as Identity from "@lucas-barake/effect-local/Identity"
 import * as ReplicaError from "@lucas-barake/effect-local/ReplicaError"
+import * as Cause from "effect/Cause"
 import * as Context from "effect/Context"
 import * as Crypto from "effect/Crypto"
 import * as DateTime from "effect/DateTime"
@@ -14,6 +15,7 @@ import * as Schema from "effect/Schema"
 import * as SqlClient from "effect/unstable/sql/SqlClient"
 import * as SqlSchema from "effect/unstable/sql/SqlSchema"
 import * as InternalAutomerge from "./internal/automerge.js"
+import * as InternalReplicaError from "./internal/replicaError.js"
 import * as WriterProvenance from "./internal/writerProvenance.js"
 import * as Recovery from "./Recovery.js"
 import * as ReplicaGate from "./ReplicaGate.js"
@@ -1058,7 +1060,7 @@ export const layer: Layer.Layer<
         }
         const definitionHash = yield* findDefinitionHash(undefined).pipe(
           Effect.map((row) => row.definition_hash),
-          Effect.catchTag("NoSuchElementError", () => Effect.die(new Error("Replica metadata was not initialized")))
+          Effect.catchIf(Cause.isNoSuchElementError, InternalReplicaError.metadataMissing)
         )
         // The stored schema version, not `document.version`: `recover` decodes at the version the
         // row records and does not migrate, so the value the re-rooted change carries is encoded at
@@ -1166,7 +1168,7 @@ export const layer: Layer.Layer<
           }
           const commitSequence = yield* nextCommitSequence(undefined).pipe(
             Effect.map((row) => row.commit_sequence),
-            Effect.catchTag("NoSuchElementError", () => Effect.die(new Error("Replica metadata was not initialized")))
+            Effect.catchIf(Cause.isNoSuchElementError, InternalReplicaError.metadataMissing)
           )
           yield* sql`DELETE FROM effect_local_changes WHERE document_id = ${documentId}`
           yield* sql`DELETE FROM effect_local_checkpoints WHERE document_id = ${documentId}`

packages/local-sql/src/DocumentStore.ts

@@ -4,6 +4,7 @@ import * as Identity from "@lucas-barake/effect-local/Identity"
 import type * as Mutation from "@lucas-barake/effect-local/Mutation"
 import * as ReplicaError from "@lucas-barake/effect-local/ReplicaError"
 import type * as Snapshot from "@lucas-barake/effect-local/Snapshot"
+import * as Cause from "effect/Cause"
 import * as Context from "effect/Context"
 import type * as Crypto from "effect/Crypto"
 import * as DateTime from "effect/DateTime"
@@ -14,6 +15,7 @@ import * as Schema from "effect/Schema"
 import * as SqlClient from "effect/unstable/sql/SqlClient"
 import * as SqlSchema from "effect/unstable/sql/SqlSchema"
 import * as InternalAutomerge from "./internal/automerge.js"
+import * as InternalReplicaError from "./internal/replicaError.js"
 import * as Rows from "./internal/rows.js"
 import * as Recovery from "./Recovery.js"
 import * as ReplicaGate from "./ReplicaGate.js"
@@ -93,8 +95,8 @@ export const layer: Layer.Layer<
           WHERE singleton = 1 RETURNING commit_sequence`
     })(undefined).pipe(
       Effect.map((row) => row.commit_sequence),
+      Effect.catchIf(Cause.isNoSuchElementError, InternalReplicaError.metadataMissing),
       Effect.catchTags({
-        NoSuchElementError: () => Effect.die(new Error("Replica metadata was not initialized")),
         SchemaError: (cause) =>
           Effect.fail(
             new ReplicaError.ReplicaError({
@@ -107,8 +109,8 @@ export const layer: Layer.Layer<
     )
     const currentDefinitionHash = findDefinitionHash(undefined).pipe(
       Effect.map((row) => row.definition_hash),
+      Effect.catchIf(Cause.isNoSuchElementError, InternalReplicaError.metadataMissing),
       Effect.catchTags({
-        NoSuchElementError: () => Effect.die(new Error("Replica metadata was not initialized")),
         SchemaError: (cause) =>
           Effect.fail(
             new ReplicaError.ReplicaError({

packages/local-sql/src/EntityReplica.ts

@@ -97,7 +97,7 @@ export const layer = (definition: ReplicaDefinition.Any): Layer.Layer<
                 })
               })
             ),
-            Effect.flatMap((lock) => lock.withPermit(f(permit)))
+            Effect.flatMap((lock) => lock.withPermit(gate.preflight(permit).pipe(Effect.andThen(f(permit)))))
           )
         )
 

packages/local-sql/src/PeerSession.ts

@@ -453,6 +453,7 @@ const makeWithTerminal = (
           Effect.gen(function*() {
             const incarnation = yield* Effect.scoped(Effect.gen(function*() {
               const permit = yield* gate.shared
+              yield* gate.preflight(permit)
               if (permit.incarnation !== session.replicaIncarnation) {
                 return yield* new ReplicaError.ReplicaError({
                   reason: new ReplicaError.ProtocolMismatch({