diff --git a/.gitignore b/.gitignore index efe1267f..fd4e98e5 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,6 @@ example # Nix /.direnv/ + +# Nix build symlinks +/result diff --git a/.prettierignore b/.prettierignore index 1bfe8454..265f57c7 100644 --- a/.prettierignore +++ b/.prettierignore @@ -3,3 +3,8 @@ node_modules .claude protobuf/gen testUtil/fixtures/gen +verification/p/PGenerated/ +verification/p/PCheckerOutput/ +verification/p/PObs/PGenerated/ +verification/p/verified/PGenerated/ +flake.lock diff --git a/PROTOCOL.md b/PROTOCOL.md index 453b50e2..602efb5a 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -235,6 +235,10 @@ interface ControlHandshakeRequest { expectedSessionState: { nextExpectedSeq: number; // integer nextSentSeq: number; // integer + // whether the client considers this a reconnection to a session that was + // previously connected. Optional for wire compatibility; servers MUST + // treat an absent flag as `false`. + isReconnect?: boolean; }; metadata?: unknown; } @@ -626,8 +630,11 @@ The server will send an error response if either: - the client wanted a reconnection to a specific session but the server doesn't know about it - the client is in the future (`client.nextSentSeq > server.ack`) - server is in the future (`server.seq > client.nextExpectedSeq`) + - the client marked the handshake as a reconnection (`isReconnect: true`) but the server has no session for it. The explicit flag matters in the _zero-state window_: a client that has sent messages but never received anything back still has `nextSentSeq: 0, nextExpectedSeq: 0`, which is otherwise indistinguishable from a brand-new session. Without the flag, a server that lost the session (restart or grace expiry) would accept such a reconnect as a new session, the client would replay its send buffer believing the reconnect was transparent, and handlers that already processed those messages would execute them a second time — while the original callers never learn anything went wrong. Rejecting instead yields the normal hard-reconnect semantics: the client starts a fresh session and in-flight calls resolve with `UNEXPECTED_DISCONNECT`. -When the client receives a status with `ok: false`, it should consider the handshake failed and close the connection. +When the client receives a status with `ok: false`, it should consider the handshake failed and close the connection. For the retriable code (`SESSION_STATE_MISMATCH`) the client MAY automatically reconnect, but MUST do so with a **fresh session** (a new session id and zeroed session state), resolving any in-flight calls of the old session with `UNEXPECTED_DISCONNECT`; retrying the same session would be rejected identically forever. For fatal codes the client MUST NOT reconnect automatically. + +Handshakes are **connection-scoped**: a handshake request or response is only meaningful on the connection that carried it. A client MUST ignore a handshake response that does not belong to its current connection attempt, and a server MUST bound the lifetime of un-handshaken connections (`handshakeTimeoutMs`), so a handshake request cannot outlive its connection. This scoping is load-bearing for the `isReconnect` guard: its correctness argument relies on a `isReconnect: false` request never being processed after the session it names has connected and transferred data (see `verification/p/verified/SessionReconnect.p`, where dropping this assumption breaks the no-duplicate-delivery proof). ### Re-handshaking (live credential refresh) diff --git a/__tests__/properties/session.property.test.ts b/__tests__/properties/session.property.test.ts index 5e3a8496..45bb89f0 100644 --- a/__tests__/properties/session.property.test.ts +++ b/__tests__/properties/session.property.test.ts @@ -20,6 +20,7 @@ import type { Connection } from '../../transport/connection'; import type { ServerTransport } from '../../transport/server'; import { closeAllConnections, numberOfConnections } from '../../testUtil'; import { createMockTransportNetwork } from '../../testUtil/fixtures/mockTransport'; +import { traceLogFn, traceSideOf } from '../../testUtil/fixtures/trace'; import type { TestTransportOptions } from '../../testUtil/fixtures/transports'; import { advanceFakeTimersByConnectionBackoff, @@ -169,11 +170,7 @@ function setup(opts?: TestTransportOptions): { const violations: Array = []; for (const t of [clientTransport, serverTransport]) { - t.bindLogger((msg, ctx, level) => { - if (ctx?.tags?.includes('invariant-violation')) { - violations.push(`[${level}] ${msg}`); - } - }, 'debug'); + t.bindLogger(traceLogFn(traceSideOf(t.clientId), violations), 'debug'); } createServer(serverTransport, services); @@ -590,11 +587,10 @@ describe('re-handshake under faults', () => { const violations: Array = []; for (const t of [clientTransport, serverTransport]) { - t.bindLogger((msg, ctx, level) => { - if (ctx?.tags?.includes('invariant-violation')) { - violations.push(`[${level}] ${msg}`); - } - }, 'debug'); + t.bindLogger( + traceLogFn(traceSideOf(t.clientId), violations), + 'debug', + ); } createServer(serverTransport, metadataServices); diff --git a/__tests__/properties/streams.property.test.ts b/__tests__/properties/streams.property.test.ts index 77dbd00a..e7eeeb51 100644 --- a/__tests__/properties/streams.property.test.ts +++ b/__tests__/properties/streams.property.test.ts @@ -11,6 +11,7 @@ import { WritableImpl, } from '../../router/streams'; import { createMockTransportNetwork } from '../../testUtil/fixtures/mockTransport'; +import { traceLogFn, traceSideOf } from '../../testUtil/fixtures/trace'; import { cleanupTransports } from '../../testUtil/fixtures/cleanup'; import type { ProvidedClientTransportOptions, @@ -117,11 +118,7 @@ async function withNetwork( const violations: Array = []; for (const t of [clientTransport, serverTransport]) { - t.bindLogger((msg, ctx, level) => { - if (ctx?.tags?.includes('invariant-violation')) { - violations.push(`[${level}] ${msg}`); - } - }, 'debug'); + t.bindLogger(traceLogFn(traceSideOf(t.clientId), violations), 'debug'); } createServer(serverTransport, services); diff --git a/__tests__/zerostate.test.ts b/__tests__/zerostate.test.ts new file mode 100644 index 00000000..a2d24fc5 --- /dev/null +++ b/__tests__/zerostate.test.ts @@ -0,0 +1,126 @@ +import { describe, expect, test } from 'vitest'; +import { Type } from 'typebox'; +import { + Ok, + Procedure, + UNEXPECTED_DISCONNECT_CODE, + createServiceSchema, +} from '../router'; +import { createClient } from '../router/client'; +import { createServer } from '../router/server'; +import { createMockTransportNetwork } from '../testUtil/fixtures/mockTransport'; +import { traceLogFn, traceSideOf } from '../testUtil/fixtures/trace'; +import { + advanceFakeTimersByConnectionBackoff, + cleanupTransports, + waitFor, +} from '../testUtil/fixtures/cleanup'; + +/** + * The "zero-state window": a client that has sent messages the server accepted + * and DELIVERED TO HANDLERS, but that has received nothing back (no response, + * no heartbeat -- so its session still reads `nextSentSeq: 0, + * nextExpectedSeq: 0`), reconnects after the server lost the session (restart + * or grace expiry). Such a handshake is indistinguishable from a brand-new + * session by the seq counters alone, so without an explicit reconnect marker + * the server accepts it as new, the client replays its send buffer, and the + * handler executes the same request a second time -- while the original call + * never resolves. + * + * This scenario was found by the P model of the protocol + * (verification/p/README.md, finding 1). The expected behavior asserted here + * is that of a client that marks reconnection attempts: the server rejects + * the unknown session with SESSION_STATE_MISMATCH, the client starts a fresh + * session, and the in-flight call resolves with UNEXPECTED_DISCONNECT -- + * exactly the documented hard-reconnect semantics, and never a duplicate + * handler execution. + */ +describe('zero-state reconnect to a server that lost the session', () => { + test('does not re-execute handlers; in-flight calls resolve with UNEXPECTED_DISCONNECT', async () => { + const invocations: Array = []; + + const ServiceSchema = createServiceSchema(); + const ZeroStateService = ServiceSchema.define({ + work: Procedure.rpc({ + requestInit: Type.Object({ id: Type.String() }), + responseData: Type.Object({}), + async handler({ ctx, reqInit }) { + invocations.push(reqInit.id); + // hang until abort so nothing (response or ack) ever flows back to + // the client, keeping the client's session in the zero-state window + await new Promise((resolve) => { + ctx.signal.addEventListener('abort', () => { + resolve(); + }); + }); + + return Ok({}); + }, + }), + }); + const services = { svc: ZeroStateService }; + + // long heartbeat interval: a server heartbeat would ack the request and + // take the client out of the zero-state window, masking the scenario + const quietHeartbeats = { + heartbeatIntervalMs: 60_000, + heartbeatsUntilDead: 2, + }; + const network = createMockTransportNetwork({ + client: { + ...quietHeartbeats, + maxJitterMs: 0, + baseIntervalMs: 10, + attemptBudgetCapacity: 100, + }, + server: quietHeartbeats, + }); + + const clientTransport = network.getClientTransport('client'); + const serverTransport = network.getServerTransport('SERVER'); + const violations: Array = []; + for (const t of [clientTransport, serverTransport]) { + t.bindLogger(traceLogFn(traceSideOf(t.clientId), violations), 'debug'); + } + + createServer(serverTransport, services); + const client = createClient(clientTransport, 'SERVER'); + + try { + // the handler runs on the first server, but the client hears nothing + const pending = client.svc.work.rpc({ id: 'once' }); + await waitFor(() => expect(invocations).toStrictEqual(['once'])); + + // the server loses all state; the client's session (and its send + // buffer holding the request) survives within its grace period + await network.restartServer(); + const secondServer = network.getServerTransport('SERVER'); + secondServer.bindLogger(traceLogFn('server', violations), 'debug'); + createServer(secondServer, services); + + await advanceFakeTimersByConnectionBackoff(); + + // the reconnect must be treated as a hard reconnect, not a fresh + // session: the caller learns its call died... + const result = await pending; + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.payload.code).toBe(UNEXPECTED_DISCONNECT_CODE); + } + + // ...and the handler must never have executed the same request twice + expect(invocations).toStrictEqual(['once']); + + // the fresh session works: new calls reach the new server + const again = client.svc.work.rpc({ id: 'later' }); + await waitFor(() => expect(invocations).toStrictEqual(['once', 'later'])); + clientTransport.hardDisconnect(); + await again; + + expect(violations).toStrictEqual([]); + } finally { + await cleanupTransports([clientTransport, serverTransport]); + await network.cleanup(); + } + }); +}); diff --git a/flake.lock b/flake.lock index 9f879ba6..92cb0357 100644 --- a/flake.lock +++ b/flake.lock @@ -1,5 +1,20 @@ { "nodes": { + "flake-utils": { + "locked": { + "lastModified": 1667395993, + "narHash": "sha256-nuEHfE/LcWyuSWnS8t12N1wc105Qtau+/OdUAjtQ0rA=", + "owner": "numtide", + "repo": "flake-utils", + "rev": "5aed5285a952e0b949eb3ba02c12fa4fcfef535f", + "type": "github" + }, + "original": { + "owner": "numtide", + "repo": "flake-utils", + "type": "github" + } + }, "nixpkgs": { "locked": { "lastModified": 1716479278, @@ -17,7 +32,29 @@ }, "root": { "inputs": { - "nixpkgs": "nixpkgs" + "nixpkgs": "nixpkgs", + "sbt-derivation": "sbt-derivation" + } + }, + "sbt-derivation": { + "inputs": { + "flake-utils": "flake-utils", + "nixpkgs": [ + "nixpkgs" + ] + }, + "locked": { + "lastModified": 1698464090, + "narHash": "sha256-Pnej7WZIPomYWg8f/CZ65sfW85IfIUjYhphMMg7/LT0=", + "owner": "zaninime", + "repo": "sbt-derivation", + "rev": "6762cf2c31de50efd9ff905cbcc87239995a4ef9", + "type": "github" + }, + "original": { + "owner": "zaninime", + "repo": "sbt-derivation", + "type": "github" } } }, diff --git a/flake.nix b/flake.nix index eeca0307..92d011b2 100644 --- a/flake.nix +++ b/flake.nix @@ -3,7 +3,13 @@ inputs.nixpkgs.url = "github:nixos/nixpkgs"; - outputs = { self, nixpkgs }: + # sbt builder used to package UCLID5 (the PVerifier proof backend). + inputs.sbt-derivation = { + url = "github:zaninime/sbt-derivation"; + inputs.nixpkgs.follows = "nixpkgs"; + }; + + outputs = { self, nixpkgs, sbt-derivation }: let mkDevShell = system: let @@ -15,11 +21,174 @@ nodePackages.typescript-language-server ]; }; + + # The P model checker (github.com/p-org/P), used by verification/p. + mkP = system: + let + pkgs = nixpkgs.legacyPackages.${system}; + in + pkgs.buildDotnetGlobalTool { + pname = "p"; + version = "3.1.0"; + nugetSha256 = "sha256-sqIS47GvG/L9ybgImdopAdZiXRouR41HjjACiHKkvcE="; + dotnet-sdk = pkgs.dotnet-sdk_8; + dotnet-runtime = pkgs.dotnet-sdk_8; + meta.mainProgram = "p"; + }; + + # UCLID5, the SMT-based verifier that `p compile --mode verification` + # (PVerifier) shells out to. Only prebuilt from source: the released + # uclid 0.9.5 zip predates the `datatype` syntax PVerifier emits, so we + # pin a recent master commit. uclid expects the z3 4.12 Java bindings as + # an unmanaged jar in lib/ (see its get-z3-linux.sh) and `z3` + the JNI + # library at runtime. + mkUclid = system: + let + pkgs = nixpkgs.legacyPackages.${system}; + z3j = pkgs.z3_4_12.override { javaBindings = true; jdk = pkgs.jdk11; }; + in + sbt-derivation.lib.mkSbtDerivation { + inherit pkgs; + # uclid pins sbt 1.4.9, whose launcher needs a pre-Security-Manager- + # removal JDK + overrides.sbt = pkgs.sbt.override { jre = pkgs.jdk11; }; + pname = "uclid"; + version = "unstable-2025-07-09"; + src = pkgs.fetchFromGitHub { + owner = "uclid-org"; + repo = "uclid"; + rev = "a4e4e7a22780833c86d6a650af2fcf88a7d00a06"; + hash = "sha256-c9oUaHcNbICJC1oWYlhE86dZtK+WaW2ZtotDWW9l2U4="; + }; + depsSha256 = "sha256-Zxl06xUzOOp1elaFvrQjKYERjCXZXB/A891/QuqEFA0="; + nativeBuildInputs = [ pkgs.makeWrapper ]; + buildPhase = '' + mkdir -p lib + cp ${z3j.java}/share/java/com.microsoft.z3.jar lib/ + sbt stage + ''; + installPhase = '' + mkdir -p $out + cp -r target/universal/stage/* $out/ + wrapProgram $out/bin/uclid \ + --prefix PATH : ${pkgs.jre}/bin \ + --prefix PATH : ${pkgs.z3_4_12}/bin \ + --prefix LD_LIBRARY_PATH : ${z3j.lib}/lib + ''; + meta.mainProgram = "uclid"; + }; + + # PObserve: P's runtime-monitoring framework (compiled P spec machines + # checked against real execution logs). Not published to Maven Central, so + # we build the three plain-Java modules (Commons, RegressionTesting, the + # CLI) straight from the P repo sources with javac — their gradle files + # only add linting/publishing plugins. Lombok runs as an annotation + # processor. The dependency jars are pinned individually from Maven + # Central. + mkPObserve = system: + let + pkgs = nixpkgs.legacyPackages.${system}; + fetchJar = name: url: hash: pkgs.fetchurl { inherit name url hash; }; + depJars = [ + (fetchJar "jcommander-1.82.jar" "https://repo1.maven.org/maven2/com/beust/jcommander/1.82/jcommander-1.82.jar" "sha256-3urBV8jeaCKHjYXQx7yEZ6GcyEhNN3iPeATwOd3igLE=") + (fetchJar "jackson-annotations-2.16.1.jar" "https://repo1.maven.org/maven2/com/fasterxml/jackson/core/jackson-annotations/2.16.1/jackson-annotations-2.16.1.jar" "sha256-pHMHceakld03k6Qs24zmvduWx34V9AyY/Y2aeuCecoY=") + (fetchJar "jackson-core-2.16.1.jar" "https://repo1.maven.org/maven2/com/fasterxml/jackson/core/jackson-core/2.16.1/jackson-core-2.16.1.jar" "sha256-9fjvkGCeZP7ILrkI5JfcfYGy65g/5Qm4cCkqGTzeTfs=") + (fetchJar "jackson-databind-2.16.1.jar" "https://repo1.maven.org/maven2/com/fasterxml/jackson/core/jackson-databind/2.16.1/jackson-databind-2.16.1.jar" "sha256-uvio6+6PRe9ozdXi3Tkjs+KWwJN7luwLSAaqOjG8zR0=") + (fetchJar "findbugs-annotations-3.0.1.jar" "https://repo1.maven.org/maven2/com/google/code/findbugs/annotations/3.0.1/annotations-3.0.1.jar" "sha256-a0f/Cm3gzhfL7cOrsIKMpbzjAJ1T6kezcj/wI8R0L3k=") + (fetchJar "log4j-api-2.20.0.jar" "https://repo1.maven.org/maven2/org/apache/logging/log4j/log4j-api/2.20.0/log4j-api-2.20.0.jar" "sha256-L0PupnnqZvFMoPE/7CqGAKwST1pSMdy034OT7dy5dVA=") + (fetchJar "log4j-core-2.20.0.jar" "https://repo1.maven.org/maven2/org/apache/logging/log4j/log4j-core/2.20.0/log4j-core-2.20.0.jar" "sha256-YTffhIza7Z9NUHb3VRPGyF2oC5U/TnrMo4CYt3B2P1U=") + (fetchJar "json-simple-1.1.1.jar" "https://repo1.maven.org/maven2/com/googlecode/json-simple/json-simple/1.1.1/json-simple-1.1.1.jar" "sha256-TmlpaJK4i0HFXUmrL9zCHurZK/VKzFiMAFBZbDt1GZw=") + (fetchJar "fst-2.50.jar" "https://repo1.maven.org/maven2/de/ruedigermoeller/fst/2.50/fst-2.50.jar" "sha256-tq5bkNWCUHDS+CrGFBcJD86O1+VjSKPoaiAmd06LpWo=") + (fetchJar "objenesis-2.5.1.jar" "https://repo1.maven.org/maven2/org/objenesis/objenesis/2.5.1/objenesis-2.5.1.jar" "sha256-sEPwPkZnUvfwPiMmo7E6SbfGSfjyotyHcVgn4k9z2cY=") + (fetchJar "javassist-3.21.0-GA.jar" "https://repo1.maven.org/maven2/org/javassist/javassist/3.21.0-GA/javassist-3.21.0-GA.jar" "sha256-eqWeAx+UGYSvB9rMbKhebcm9OkhemqJJTLwDTvoSJdA=") + (fetchJar "awssdk-annotations-2.29.15.jar" "https://repo1.maven.org/maven2/software/amazon/awssdk/annotations/2.29.15/annotations-2.29.15.jar" "sha256-H9gLfpoqFOjRsCbpDYaLhiopuxXtmiLyAnmb+8rRgH4=") + (fetchJar "awssdk-regions-2.29.15.jar" "https://repo1.maven.org/maven2/software/amazon/awssdk/regions/2.29.15/regions-2.29.15.jar" "sha256-vg8V7XPtFx97yM3mh3lpcI11A/QLj+WfhteqlZqoG8c=") + (fetchJar "awssdk-utils-2.29.15.jar" "https://repo1.maven.org/maven2/software/amazon/awssdk/utils/2.29.15/utils-2.29.15.jar" "sha256-XP6qaDlVOF9O3FHqj7sydWsPpaimtSNw6L0JN9uSvoY=") + (fetchJar "validation-api-2.0.1.Final.jar" "https://repo1.maven.org/maven2/javax/validation/validation-api/2.0.1.Final/validation-api-2.0.1.Final.jar" "sha256-mHO0bfGDPJ7o9bwf9oUzdRFdrdiJe8taDf+1hIg17mw=") + ]; + lombok = fetchJar "lombok-1.18.30.jar" "https://repo1.maven.org/maven2/org/projectlombok/lombok/1.18.30/lombok-1.18.30.jar" "sha256-FBUbR1gtVwtN4WoUfs4729GazkruW946VXjIfbnsuZg="; + in + pkgs.stdenv.mkDerivation { + pname = "pobserve"; + version = "1.0.0-p3.1.0"; + src = pkgs.fetchFromGitHub { + owner = "p-org"; + repo = "P"; + rev = "857776015de5f2683f11945728a1dc57f4b74b33"; # the P 3.1.0 release commit + hash = "sha256-5YJ+YY20pqNKd4Z7tuqa8+HAHSG1e2NyqqAdyf0ExQ0="; + }; + nativeBuildInputs = [ pkgs.jdk17 pkgs.makeWrapper pkgs.strip-nondeterminism ]; + buildPhase = '' + runHook preBuild + cp ${lombok} lombok.jar + depcp=$(echo ${pkgs.lib.concatMapStringsSep ":" toString depJars}) + mkdir -p classes + find Src/PObserve/PObserveCommons/src/main/java Src/PObserve/PObserve/src/main/java -name '*.java' > sources.txt + javac -cp "$depcp:lombok.jar" -processorpath lombok.jar -d classes @sources.txt + cp -r Src/PObserve/PObserve/src/main/resources/* classes/ 2>/dev/null || true + jar cf pobserve.jar -C classes . + runHook postBuild + ''; + installPhase = '' + runHook preInstall + mkdir -p $out/share/java $out/bin + cp pobserve.jar $out/share/java/ + strip-nondeterminism --type jar $out/share/java/pobserve.jar || true + depcp=$(echo ${pkgs.lib.concatMapStringsSep ":" toString depJars}) + makeWrapper ${pkgs.jdk17}/bin/java $out/bin/pobserve --add-flags "-cp $out/share/java/pobserve.jar:$depcp pobserve.PObserve" + # the compile classpath, for building spec/parser jars against pobserve + echo "$out/share/java/pobserve.jar:$depcp" > $out/share/java/classpath.txt + runHook postInstall + ''; + meta.mainProgram = "pobserve"; + }; + + # `p compile`/`p check` shell out to `dotnet build` at runtime, so the + # SDK itself must be on PATH alongside the tool. JDK 17 + Maven power the + # PEx exhaustive-checking backend (`p compile --mode pex`); uclid powers + # the PVerifier proof backend (`p compile --mode verification`). + mkVerificationShell = system: + let + pkgs = nixpkgs.legacyPackages.${system}; + in + pkgs.mkShell { + nativeBuildInputs = [ + (mkP system) + (mkUclid system) + (mkPObserve system) + pkgs.dotnet-sdk_8 + pkgs.jdk17 + pkgs.maven + pkgs.kotlin + ]; + POBSERVE_HOME = mkPObserve system; + DOTNET_CLI_TELEMETRY_OPTOUT = "1"; + }; in { devShells.aarch64-linux.default = mkDevShell "aarch64-linux"; devShells.aarch64-darwin.default = mkDevShell "aarch64-darwin"; devShells.x86_64-linux.default = mkDevShell "x86_64-linux"; devShells.x86_64-darwin.default = mkDevShell "x86_64-darwin"; + + packages.aarch64-linux.p = mkP "aarch64-linux"; + packages.aarch64-darwin.p = mkP "aarch64-darwin"; + packages.x86_64-linux.p = mkP "x86_64-linux"; + packages.x86_64-darwin.p = mkP "x86_64-darwin"; + + packages.aarch64-linux.pobserve = mkPObserve "aarch64-linux"; + packages.aarch64-darwin.pobserve = mkPObserve "aarch64-darwin"; + packages.x86_64-linux.pobserve = mkPObserve "x86_64-linux"; + packages.x86_64-darwin.pobserve = mkPObserve "x86_64-darwin"; + + packages.aarch64-linux.uclid = mkUclid "aarch64-linux"; + packages.aarch64-darwin.uclid = mkUclid "aarch64-darwin"; + packages.x86_64-linux.uclid = mkUclid "x86_64-linux"; + packages.x86_64-darwin.uclid = mkUclid "x86_64-darwin"; + + devShells.aarch64-linux.verification = mkVerificationShell "aarch64-linux"; + devShells.aarch64-darwin.verification = mkVerificationShell "aarch64-darwin"; + devShells.x86_64-linux.verification = mkVerificationShell "x86_64-linux"; + devShells.x86_64-darwin.verification = mkVerificationShell "x86_64-darwin"; }; } diff --git a/package.json b/package.json index 16511c70..79d43bd9 100644 --- a/package.json +++ b/package.json @@ -71,7 +71,9 @@ "test": "vitest", "test:single": "vitest run --reporter=dot", "test:flake": "./flake.sh", - "bench": "vitest bench" + "bench": "vitest bench", + "model:check": "bash verification/p/check.sh", + "model:observe": "bash verification/p/observe.sh" }, "engines": { "node": ">=20.11.0" diff --git a/testUtil/fixtures/mockTransport.ts b/testUtil/fixtures/mockTransport.ts index fd7369f9..a10ab83e 100644 --- a/testUtil/fixtures/mockTransport.ts +++ b/testUtil/fixtures/mockTransport.ts @@ -14,6 +14,13 @@ import { ClientHandshakeOptions, ServerHandshakeOptions, } from '../../router/handshake'; +import { NaiveJsonCodec } from '../../codec'; +import { + attachTraceEvents, + startTraceCase, + traceWrapCodec, + tracingEnabled, +} from './trace'; export class InMemoryConnection extends Connection { conn: Duplex; @@ -72,6 +79,10 @@ interface BidiConnection { export function createMockTransportNetwork( opts?: TestTransportOptions, ): TestSetupHelpers { + // one trace file per network = per generated test case (no-op unless + // RIVER_TRACE_DIR is set -- see ./trace.ts) + startTraceCase(); + // conn id -> [client->server, server->client] const connections = new Observable>({}); @@ -151,12 +162,21 @@ export function createMockTransportNetwork( ) => { const clientTransport = new MockClientTransport( id, - opts?.client, + tracingEnabled() + ? { + ...opts?.client, + codec: traceWrapCodec( + 'client', + opts?.client?.codec ?? NaiveJsonCodec, + ), + } + : opts?.client, ); if (handshakeOptions) { clientTransport.extendHandshake(handshakeOptions); } + attachTraceEvents('client', clientTransport); transports.push(clientTransport); return clientTransport; @@ -179,11 +199,23 @@ export function createMockTransportNetwork( MetadataSchema, ParsedMetadata, RejectionCodeSchema - >(id, opts?.server); + >( + id, + tracingEnabled() + ? { + ...opts?.server, + codec: traceWrapCodec( + 'server', + opts?.server?.codec ?? NaiveJsonCodec, + ), + } + : opts?.server, + ); if (handshakeOptions) { serverTransport.extendHandshake(handshakeOptions); } + attachTraceEvents('server', serverTransport); transports.push(serverTransport); return serverTransport; diff --git a/testUtil/fixtures/trace.ts b/testUtil/fixtures/trace.ts new file mode 100644 index 00000000..32455db4 --- /dev/null +++ b/testUtil/fixtures/trace.ts @@ -0,0 +1,162 @@ +import { appendFileSync, mkdirSync } from 'node:fs'; +import { join } from 'node:path'; +import type { Codec } from '../../codec'; +import type { Connection } from '../../transport/connection'; +import type { Transport } from '../../transport/transport'; +import type { LogFn, MessageMetadata } from '../../logging/log'; +import type { + OpaqueTransportMessage, + TransportClientId, +} from '../../transport/message'; + +/** + * Test-only execution tracing for runtime conformance checking against the P + * model (verification/p/PObs, run via PObserve — see + * verification/p/observe.sh). + * + * Enabled by setting RIVER_TRACE_DIR; otherwise every hook is a no-op. When + * enabled, each mock network (i.e. each generated property-test case) writes + * one JSONL file of transport-level records: encoded/accepted frames with + * seq/ack/flags, session lifecycle events, and the transport's own + * invariant-violation log lines. Records carry a process-monotonic counter + * `n` as the ordering key (the test clock is faked and jumps). + */ + +type TraceSide = 'client' | 'server'; + +const traceDir = process.env.RIVER_TRACE_DIR; +let counter = 0; +let caseSeq = 0; +let currentFile: string | undefined; +let currentRun = ''; + +export function tracingEnabled(): boolean { + return traceDir !== undefined && traceDir !== ''; +} + +/** Start a fresh trace file. Called once per mock network construction. */ +export function startTraceCase(): void { + if (!traceDir) return; + mkdirSync(traceDir, { recursive: true }); + currentRun = `${process.pid}-${caseSeq++}`; + currentFile = join(traceDir, `${currentRun}.jsonl`); +} + +function emit(rec: Record): void { + if (!traceDir || !currentFile) return; + appendFileSync( + currentFile, + `${JSON.stringify({ n: counter++, run: currentRun, ...rec })}\n`, + ); +} + +function isHandshakePayload(payload: unknown): boolean { + return ( + typeof payload === 'object' && + payload !== null && + 'type' in payload && + ((payload as { type: unknown }).type === 'HANDSHAKE_REQ' || + (payload as { type: unknown }).type === 'HANDSHAKE_RESP') + ); +} + +/** + * Wrap a codec so every outbound frame is recorded at encode time. + * Handshake frames are out-of-band (seq 0/ack 0) and skipped. Retransmits + * reuse the cached encoding and deliberately do not re-appear here. + */ +export function traceWrapCodec(side: TraceSide, inner: Codec): Codec { + if (!tracingEnabled()) return inner; + + return { + toBuffer(msg) { + const m = msg as Partial; + if (!isHandshakePayload(m.payload) && typeof m.seq === 'number') { + emit({ + side, + k: 'enc', + seq: m.seq, + ack: m.ack, + streamId: m.streamId, + controlFlags: m.controlFlags, + }); + } + + return inner.toBuffer(msg); + }, + fromBuffer: (buf) => inner.fromBuffer(buf), + }; +} + +/** Subscribe to a transport's lifecycle events. */ +export function attachTraceEvents( + side: TraceSide, + transport: Transport, +): void { + if (!tracingEnabled()) return; + + transport.addEventListener('sessionStatus', (evt) => { + // snapshot synchronously: session handles throw once consumed + if (evt.status === 'closed') { + emit({ side, k: 'sclosed', sessionId: evt.session.id }); + + return; + } + + emit({ + side, + k: evt.status === 'created' ? 'screate' : 'sclosing', + sessionId: evt.session.id, + state: evt.session.state, + }); + }); + transport.addEventListener('sessionTransition', (evt) => { + emit({ side, k: 'strans', sessionId: evt.id, state: evt.state }); + }); + transport.addEventListener('protocolError', (evt) => { + emit({ side, k: 'perr', type: evt.type, message: evt.message }); + }); +} + +/** + * A log function that records the transport's per-message narrative + * (accepted frames, out-of-order receives, invariant violations) and also + * collects `invariant-violation` lines the way the property tests expect. + */ +export function traceLogFn(side: TraceSide, violations: Array): LogFn { + return (msg: string, ctx?: MessageMetadata, level?: string) => { + if (ctx?.tags?.includes('invariant-violation')) { + violations.push(`[${level ?? '?'}] ${msg}`); + emit({ side, k: 'inv', message: msg }); + } + if (!tracingEnabled()) return; + + if (msg === 'received msg' && ctx?.transportMessage) { + const m = ctx.transportMessage; + emit({ + side, + k: 'acc', + seq: m.seq, + ack: m.ack, + streamId: m.streamId, + controlFlags: m.controlFlags, + sessionId: ctx.sessionId, + }); + } else if (msg.startsWith('received out-of-order msg')) { + const m = ctx?.transportMessage; + emit({ + side, + k: 'ooo', + seq: m?.seq, + ack: m?.ack, + streamId: m?.streamId, + sessionId: ctx?.sessionId, + }); + } + }; +} + +/** Which side a transport is, by the property suite's naming convention. */ +export function traceSideOf(clientId: TransportClientId): TraceSide { + return clientId === 'SERVER' ? 'server' : 'client'; +} diff --git a/transport/client.ts b/transport/client.ts index 83e1a9b4..acdb2d18 100644 --- a/transport/client.ts +++ b/transport/client.ts @@ -734,6 +734,7 @@ export abstract class ClientTransport< expectedSessionState: { nextExpectedSeq: session.ack, nextSentSeq: session.nextSeq(), + isReconnect: session.hadConnection, }, metadata, tracing: getPropagationContext(session.telemetry.ctx), diff --git a/transport/message.ts b/transport/message.ts index e06e4aac..7c931363 100644 --- a/transport/message.ts +++ b/transport/message.ts @@ -98,6 +98,16 @@ export const ControlMessageHandshakeRequestSchema = Type.Object({ // what the client expects the server to send next nextExpectedSeq: Type.Integer(), nextSentSeq: Type.Integer(), + /** + * Whether the client considers this a reconnection to a session that was + * previously connected. Lets the server reject a reconnect to a session + * it has lost even when both seq counters are zero (the client sent + * messages but never received anything back) -- otherwise such a + * handshake is indistinguishable from a brand-new session and the + * client's send-buffer replay would re-execute handlers. Optional for + * wire compatibility: servers treat an absent flag as `false`. + */ + isReconnect: Type.Optional(Type.Boolean()), }), metadata: Type.Optional(Type.Unknown()), diff --git a/transport/server.ts b/transport/server.ts index 6f69fdc2..bc4c4aa4 100644 --- a/transport/server.ts +++ b/transport/server.ts @@ -650,7 +650,12 @@ export abstract class ServerTransport< oldSession = undefined; } - if (!oldSession && (clientNextSentSeq > 0 || clientNextExpectedSeq > 0)) { + if ( + !oldSession && + (clientNextSentSeq > 0 || + clientNextExpectedSeq > 0 || + msg.payload.expectedSessionState.isReconnect === true) + ) { // we don't have a session, but the client is trying to reconnect // to an old session. we can't do anything about this, so we reject connectCase = 'unknown session'; diff --git a/transport/sessionStateMachine/common.ts b/transport/sessionStateMachine/common.ts index 96a06607..0f30b8c1 100644 --- a/transport/sessionStateMachine/common.ts +++ b/transport/sessionStateMachine/common.ts @@ -190,6 +190,7 @@ export type InheritedProperties = Pick< | 'seqSent' | 'sendBuffer' | 'sendBufferDrainWaiter' + | 'hadConnection' | 'telemetry' | 'options' >; @@ -212,6 +213,7 @@ export interface IdentifiedSessionProps extends CommonSessionProps { seqSent: number; sendBuffer: Array; sendBufferDrainWaiter: PromiseWithResolvers | undefined; + hadConnection: boolean; telemetry: TelemetryInfo; protocolVersion: ProtocolVersion; listeners: IdentifiedSessionListeners; @@ -240,6 +242,17 @@ export abstract class IdentifiedSession extends CommonSession { ack: number; sendBuffer: Array; + /** + * Whether this session has ever reached the Connected state. Sent as + * `expectedSessionState.isReconnect` in handshake requests so the server + * can distinguish a reconnection attempt from a brand-new session even + * when the seq/ack counters are still all-zero (nothing was ever acked + * back). Without it, a reconnect to a server that lost the session in + * that window is accepted as a new session and the send-buffer replay + * re-executes handlers. + */ + hadConnection: boolean; + /** * Shared promise for pending {@link waitForSendBufferDrain} calls, created * lazily on the first waiter of a pressure episode and cleared on drain. @@ -255,6 +268,7 @@ export abstract class IdentifiedSession extends CommonSession { ack, sendBuffer, sendBufferDrainWaiter, + hadConnection, telemetry, log, protocolVersion, @@ -268,6 +282,7 @@ export abstract class IdentifiedSession extends CommonSession { this.ack = ack; this.sendBuffer = sendBuffer; this.sendBufferDrainWaiter = sendBufferDrainWaiter; + this.hadConnection = hadConnection; this.telemetry = telemetry; this.log = log; this.protocolVersion = protocolVersion; diff --git a/transport/sessionStateMachine/transitions.ts b/transport/sessionStateMachine/transitions.ts index 41df4313..71bab112 100644 --- a/transport/sessionStateMachine/transitions.ts +++ b/transport/sessionStateMachine/transitions.ts @@ -60,6 +60,7 @@ function inheritSharedSession( seqSent: session.seqSent, sendBuffer: session.sendBuffer, sendBufferDrainWaiter: session.sendBufferDrainWaiter, + hadConnection: session.hadConnection, telemetry: session.telemetry, options: session.options, log: session.log, @@ -104,6 +105,7 @@ export const SessionStateGraph = { graceExpiryTime: Date.now() + options.sessionDisconnectGraceMs, sendBuffer, sendBufferDrainWaiter: undefined, + hadConnection: false, telemetry, options, protocolVersion, @@ -238,6 +240,7 @@ export const SessionStateGraph = { conn, listeners, ...carriedState, + hadConnection: true, }); session.startHeartbeatWatchdog(); @@ -276,6 +279,7 @@ export const SessionStateGraph = { seqSent: 0, sendBuffer: [], sendBufferDrainWaiter: undefined, + hadConnection: true, telemetry: createSessionTelemetryInfo( pendingSession.tracer, sessionId, diff --git a/verification/p/.gitignore b/verification/p/.gitignore new file mode 100644 index 00000000..2b80777c --- /dev/null +++ b/verification/p/.gitignore @@ -0,0 +1,9 @@ +PGenerated/ +PCheckerOutput/ +verified/PGenerated/ +PObs/PGenerated/ +PObs/build/ +PObs/river-trace.jar +PObs/river-parser.jar +PObs/traces/ +PObs/out/ diff --git a/verification/p/PObs/RiverTraceSpecs.p b/verification/p/PObs/RiverTraceSpecs.p new file mode 100644 index 00000000..5c84abb6 --- /dev/null +++ b/verification/p/PObs/RiverTraceSpecs.p @@ -0,0 +1,211 @@ +/***************************************************************************** +Runtime-conformance specs (PObserve): the P model's invariants checked against +REAL executions of the TypeScript implementation. + +The hegel property-based suite (or any test run) emits a JSONL trace via +testUtil/fixtures/trace.ts — accepted/encoded frames with seq/ack/flags, +session lifecycle events, and invariant-violation log lines — with zero +library changes (transport events + bindLogger + a wrapper codec). The Java +parser (verification/p/pobserve-bridge) turns each line into one of the +events below; PObserve sorts them by the tap's monotonic counter and routes +them to per-partition monitor instances. + +Each spec is checked in its own PObserve run because they partition the +stream differently (see observe.sh): + + AcceptedSeqContiguous key = sessionId | receiverSide + EncodedSeqDense key = side + SessionStateConformance key = side | sessionId + StreamFlagDiscipline key = sessionId | streamId | receiverSide + NoInvariantViolations key = constant + +Because a partition IS a session (or stream), "fresh monitor instance" and +"fresh session" coincide: seq contiguity is enforced within a session across +transparent reconnects, and resets naturally with a new session id — the +same scoping as the model checker's ExactlyOnceInOrder monitor. +*****************************************************************************/ + +// An accepted or encoded transport frame, as observed at one endpoint. +// atServer: which endpoint observed it. For eTAccepted the observer is the +// RECEIVER; for eTEncoded the observer is the SENDER. +type tFrame = (atServer: bool, seqn: int, ack: int, streamId: string, + sessionId: string, isAck: bool, isOpen: bool, isClose: bool, + isCancel: bool); + +// Session lifecycle, from the transport's sessionStatus/sessionTransition +// events. state is the SessionState string ('NoConnection', 'Connecting', +// 'Handshaking', 'Connected', 'BackingOff'). +type tSessionEvt = (atServer: bool, sessionId: string, sname: string); + +event eTAccepted: tFrame; // 'received msg': passed the seq==ack gate +event eTEncoded: tFrame; // codec tap: outbound frame at encode time +event eTOutOfOrder: tFrame; // 'received out-of-order msg' (seq > ack) +event eTSessionCreated: tSessionEvt; +event eTSessionTransition: tSessionEvt; +event eTSessionClosing: tSessionEvt; +event eTInvariantViolation: (atServer: bool, message: string); + +/***************************************************************************** +C1/C2 on real traces: within one session (partition = sessionId|receiver), +accepted seqs are exactly 0, 1, 2, ... — no gap, duplicate, or reorder ever +reaches the bookkeeping layer, including across transparent reconnects +(same session id => same partition => the counter carries over). +*****************************************************************************/ +spec AcceptedSeqContiguous observes eTAccepted { + var expected: int; + start state Watching { + on eTAccepted do (f: tFrame) { + assert f.seqn == expected, + format("accepted seq {0} but expected {1} (session {2})", f.seqn, expected, f.sessionId); + expected = f.seqn + 1; + } + } +} + +/***************************************************************************** +assertSendOrdering's observable shadow: each endpoint assigns seq numbers +densely at encode time. Encode records carry no session id (the codec sits +below the session), so a reset to 0 is always allowed (a fresh session); +within a session the assignment must be +1. Partition = side. +*****************************************************************************/ +spec EncodedSeqDense observes eTEncoded { + var expected: int; + start state Watching { + on eTEncoded do (f: tFrame) { + assert f.seqn == expected || f.seqn == 0, + format("encoded seq {0} but expected {1} or 0 (new session)", f.seqn, expected); + expected = f.seqn + 1; + } + } +} + +/***************************************************************************** +The session state machine of the implementation must follow the transition +graph the model (and transitions.ts) declares. Partition = side|sessionId. + +Client edges: NoConnection->{BackingOff}, BackingOff->{Connecting}, +Connecting->{Handshaking, NoConnection}, Handshaking->{Connected, +NoConnection}, Connected->{NoConnection}. (BackingOff->NoConnection is +declared in transitions.ts but marked unused — if a real trace ever takes +it, that is a genuine model/code mismatch worth investigating.) + +Server edges: sessions are born Connected (WaitingForHandshakeToConnected), +then Connected<->NoConnection (drop / transparent reconnect adoption). +*****************************************************************************/ +spec SessionStateConformance observes eTSessionCreated, eTSessionTransition { + var cur: string; + var isServer: bool; + var started: bool; + start state Watching { + on eTSessionCreated do (s: tSessionEvt) { + assert !started, format("session {0} created twice", s.sessionId); + started = true; + isServer = s.atServer; + if (s.atServer) { + assert s.sname == "Connected", + format("server session {0} created in state {1}, expected Connected", s.sessionId, s.sname); + } else { + assert s.sname == "NoConnection", + format("client session {0} created in state {1}, expected NoConnection", s.sessionId, s.sname); + } + cur = s.sname; + } + on eTSessionTransition do (s: tSessionEvt) { + // the creation dispatch also emits a transition event for the initial + // state; accept it as a self-edge on the start state + if (started && s.sname == cur) { + return; + } + assert started, format("transition for unknown session {0}", s.sessionId); + if (isServer) { + assert legalServerEdge(cur, s.sname), + format("server session {0}: illegal transition {1} -> {2}", s.sessionId, cur, s.sname); + } else { + assert legalClientEdge(cur, s.sname), + format("client session {0}: illegal transition {1} -> {2}", s.sessionId, cur, s.sname); + } + cur = s.sname; + } + } + + fun legalClientEdge(a: string, b: string): bool { + if (a == "NoConnection") { + return b == "BackingOff"; + } + if (a == "BackingOff") { + return b == "Connecting"; + } + if (a == "Connecting") { + return b == "Handshaking" || b == "NoConnection"; + } + if (a == "Handshaking") { + return b == "Connected" || b == "NoConnection"; + } + if (a == "Connected") { + return b == "NoConnection"; + } + return false; + } + + fun legalServerEdge(a: string, b: string): bool { + if (a == "Connected") { + return b == "NoConnection"; + } + if (a == "NoConnection") { + return b == "Connected"; + } + return false; + } +} + +/***************************************************************************** +B-series flag discipline on real traces, per stream pipe +(partition = sessionId|streamId|receiverSide; reserved control streams are +filtered out by the parser): + - the request pipe's first accepted frame carries StreamOpenBit, and only + the first; + - after a sender's Close or Cancel, that sender writes nothing further on + the stream (half-close: the OTHER side may keep writing). +*****************************************************************************/ +spec StreamFlagDiscipline observes eTAccepted { + var seen: bool; + var senderClosed: bool; + start state Watching { + on eTAccepted do (f: tFrame) { + if (f.atServer) { + // request pipe: client-opened + if (!seen) { + assert f.isOpen, + format("first frame of stream {0} lacks StreamOpenBit", f.streamId); + } else { + assert !f.isOpen, + format("stream {0} opened twice", f.streamId); + } + } + assert !senderClosed, + format("frame on stream {0} after the sender closed its writer", f.streamId); + seen = true; + if (f.isClose || f.isCancel) { + senderClosed = true; + } + } + } +} + +/***************************************************************************** +C4 on real traces: the implementation's own invariant-violation log lines +(assertSendOrdering, seq > ack among honest peers, session-map invariants) +must never fire, and neither should the out-of-order receive path — the +transport under test runs over reliable in-order connections. +*****************************************************************************/ +spec NoInvariantViolations observes eTInvariantViolation, eTOutOfOrder { + start state Watching { + on eTInvariantViolation do (v: (atServer: bool, message: string)) { + assert false, format("implementation logged an invariant violation: {0}", v.message); + } + on eTOutOfOrder do (f: tFrame) { + assert false, + format("out-of-order frame reached an endpoint (seq {0}) over a reliable connection", f.seqn); + } + } +} diff --git a/verification/p/PObs/parser/RiverTraceParser.kt b/verification/p/PObs/parser/RiverTraceParser.kt new file mode 100644 index 00000000..6a130840 --- /dev/null +++ b/verification/p/PObs/parser/RiverTraceParser.kt @@ -0,0 +1,151 @@ +package river.pobserve + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.ObjectMapper +import generatedOutput.pobserve.PEvents +import generatedOutput.pobserve.PTypes +import java.util.stream.Stream +import pobserve.commons.PObserveEvent +import pobserve.commons.Parser +import pobserve.runtime.events.PEvent + +/** + * Maps the JSONL execution traces emitted by testUtil/fixtures/trace.ts to + * the P events of verification/p/PObs/RiverTraceSpecs.p. + * + * The parser is mode-selected via --parserConfiguration because each spec + * partitions the stream differently (PObserve routes one monitor instance + * per partition key): + * + * session : eTAccepted, key = sessionId|side (AcceptedSeqContiguous) + * side : eTEncoded, key = side (EncodedSeqDense) + * machine : eTSessionCreated/ + * eTSessionTransition, key = side|sessionId (SessionStateConformance) + * stream : eTAccepted minus + * control streams/acks, key = sess|stream|side (StreamFlagDiscipline) + * global : eTInvariantViolation/ + * eTOutOfOrder, key = "g" (NoInvariantViolations) + * + * The trace's process-monotonic counter `n` is the ordering timestamp. + */ +class RiverTraceParser : Parser> { + private var mode = "session" + + override fun setConfiguration(configuration: String?) { + if (!configuration.isNullOrEmpty()) { + mode = configuration + } + } + + override fun apply(logLine: Any): Stream>> { + val line = logLine.toString() + if (line.isBlank()) { + return Stream.empty() + } + val j: JsonNode = + try { + MAPPER.readTree(line) + } catch (e: Exception) { + return Stream.empty() + } ?: return Stream.empty() + if (!j.hasNonNull("k")) { + return Stream.empty() + } + + val n = j.path("n").asLong() + val k = j.path("k").asText() + // every partition key is scoped by the run (process + generated case): + // sessions/streams never span runs, and the ordering counter `n` is + // per-process + val run = j.path("run").asText("") + val rawSide = j.path("side").asText() + val side = "$run|$rawSide" + val atServer = rawSide == "server" + val sessionId = j.path("sessionId").asText("") + + var event: PEvent<*>? = null + var key: String? = null + + when (mode) { + "session" -> + if (k == "acc") { + event = PEvents.eTAccepted(frame(j, atServer, sessionId)) + key = "$sessionId|$side" + } + "side" -> + if (k == "enc") { + event = PEvents.eTEncoded(frame(j, atServer, sessionId)) + key = side + } + "machine" -> + if (k == "screate" || k == "strans") { + val s = PTypes.PTuple_atsrv_sssn_sname() + s.atServer = atServer + s.sessionId = sessionId + s.sname = j.path("state").asText("") + event = + if (k == "screate") PEvents.eTSessionCreated(s) + else PEvents.eTSessionTransition(s) + key = "$side|$sessionId" + } + "stream" -> + if (k == "acc") { + val streamId = j.path("streamId").asText("") + val flags = j.path("controlFlags").asLong(0) + val isControl = + streamId == "heartbeat" || + streamId == "rehandshake" || + (flags and ACK_BIT) != 0L + if (!isControl) { + event = PEvents.eTAccepted(frame(j, atServer, sessionId)) + key = "$sessionId|$streamId|$side" + } + } + "global" -> + if (k == "inv") { + val v = PTypes.PTuple_atsrv_mssg() + v.atServer = atServer + v.message = j.path("message").asText("") + event = PEvents.eTInvariantViolation(v) + key = "g" + } else if (k == "ooo") { + event = PEvents.eTOutOfOrder(frame(j, atServer, sessionId)) + key = "g" + } + else -> throw IllegalArgumentException("unknown parser mode: $mode") + } + + if (event == null) { + return Stream.empty() + } + + return Stream.of(PObserveEvent>(key, n, event, line)) + } + + private companion object { + private val MAPPER = ObjectMapper() + private const val ACK_BIT = 0b00001L + private const val STREAM_OPEN_BIT = 0b00010L + private const val STREAM_CANCEL_BIT = 0b00100L + private const val STREAM_CLOSED_BIT = 0b01000L + + private fun frame( + j: JsonNode, + atServer: Boolean, + sessionId: String, + ): PTypes.PTuple_atsrv_seqn_ack_strm_sssn_isAck_Open_Close_cncl { + val flags = j.path("controlFlags").asLong(0) + val f = PTypes.PTuple_atsrv_seqn_ack_strm_sssn_isAck_Open_Close_cncl() + f.atServer = atServer + f.seqn = j.path("seq").asLong(-1) + f.ack = j.path("ack").asLong(-1) + f.streamId = j.path("streamId").asText("") + f.sessionId = sessionId + f.isAck = (flags and ACK_BIT) != 0L + f.isOpen = (flags and STREAM_OPEN_BIT) != 0L + f.isClose = (flags and STREAM_CLOSED_BIT) != 0L + f.isCancel = (flags and STREAM_CANCEL_BIT) != 0L + return f + } + } +} diff --git a/verification/p/PObs/river-trace-gen.jar b/verification/p/PObs/river-trace-gen.jar new file mode 100644 index 00000000..c4ed4331 Binary files /dev/null and b/verification/p/PObs/river-trace-gen.jar differ diff --git a/verification/p/PSpec/Delivery.p b/verification/p/PSpec/Delivery.p new file mode 100644 index 00000000..89d2d166 --- /dev/null +++ b/verification/p/PSpec/Delivery.p @@ -0,0 +1,109 @@ +/***************************************************************************** +Delivery specs. + +ExactlyOnceInOrder (C1, C2 observable consequence): between session resets, +app payloads are accepted by each side exactly once, in issue order, with no +gaps or duplicates — in particular across any schedule of TRANSPARENT +reconnects, which announce nothing to this monitor. + +After an eSpecReset (a session was destroyed: grace expiry, server restart, +hard reconnect, retriable rejection) a delivery GAP is allowed — messages +die with their session and resolve UNEXPECTED_DISCONNECT — but re-delivery +of an already-delivered payload is NOT: the watermark may only move forward. + +Historical note: re-delivery across a reset used to be possible (the model +checker surfaced it, and __tests__/zerostate.test.ts reproduced it against +the TypeScript implementation): if the server delivered a message but every +ack/echo back to the client was lost, and the server then lost the session +before the client reconnected, the client's handshake presented +nextSentSeq=0/nextExpectedSeq=0 — indistinguishable from a brand-new +session — so the server accepted it (connect case 4) and the client's +buffer replay re-executed handlers. The handshake's +expectedSessionState.isReconnect flag closes that window; the model keeps +the pre-fix behavior behind zeroStateGuardFixed=false, and tcZeroStateDup +asserts the checker still FINDS the duplicate there. + +AllCallsResolve (C5 + liveness): every issued call resolves exactly once — +with a value, or with UNEXPECTED_DISCONNECT when its session died. The +monitor is hot while any call is unresolved, so an execution that quiesces +with a hanging call is a liveness bug. +*****************************************************************************/ + +spec ExactlyOnceInOrder observes eSpecDeliver, eSpecReset { + var expectedAtServer: int; + var expectedAtClient: int; + var freshAtServer: bool; // a reset happened since the last delivery + var freshAtClient: bool; + + start state Watching { + entry { + expectedAtServer = 1; + expectedAtClient = 1; + } + on eSpecDeliver do (d: (atServer: bool, payload: int)) { + if (d.atServer) { + if (freshAtServer) { + assert d.payload >= expectedAtServer, + format("server re-delivered {0} after a session reset (watermark {1})", + d.payload, expectedAtServer); + } else { + assert d.payload == expectedAtServer, + format("server delivered {0}, expected {1}, with no intervening session reset", + d.payload, expectedAtServer); + } + expectedAtServer = d.payload + 1; + freshAtServer = false; + } else { + if (freshAtClient) { + assert d.payload >= expectedAtClient, + format("client re-delivered {0} after a session reset (watermark {1})", + d.payload, expectedAtClient); + } else { + assert d.payload == expectedAtClient, + format("client delivered {0}, expected {1}, with no intervening session reset", + d.payload, expectedAtClient); + } + expectedAtClient = d.payload + 1; + freshAtClient = false; + } + } + on eSpecReset do { + freshAtServer = true; + freshAtClient = true; + } + } +} + +spec AllCallsResolve observes eSpecCall, eSpecResolved { + var pending: map[int, bool]; + var everResolved: map[int, bool]; + + start cold state AllResolved { + on eSpecCall do (p: int) { + trackCall(p); + goto Pending; + } + on eSpecResolved do (r: (payload: int, ok: bool)) { + assert false, format("call {0} resolved twice (nothing pending)", r.payload); + } + } + + hot state Pending { + on eSpecCall do (p: int) { trackCall(p); } + on eSpecResolved do (r: (payload: int, ok: bool)) { + assert r.payload in pending, + format("call {0} resolved but was not pending (double resolution?)", r.payload); + pending -= (r.payload); + everResolved[r.payload] = true; + if (sizeof(pending) == 0) { + goto AllResolved; + } + } + } + + fun trackCall(p: int) { + assert !(p in pending) && !(p in everResolved), + format("call {0} issued twice", p); + pending[p] = true; + } +} diff --git a/verification/p/PSpec/Reconnect.p b/verification/p/PSpec/Reconnect.p new file mode 100644 index 00000000..b7dfe10e --- /dev/null +++ b/verification/p/PSpec/Reconnect.p @@ -0,0 +1,55 @@ +/***************************************************************************** +Reconnect / lifecycle specs. + +SessionIdPreserved (C6): a transparent reconnect adopts the SAME session id. + +CleanShutdown (C7): once shutdown starts, both endpoints eventually announce +that they closed (sessions destroyed, connections closed). Hot until then. +*****************************************************************************/ + +spec SessionIdPreserved observes eSpecTransparentReconnect { + start state Watching { + on eSpecTransparentReconnect do (t: (oldId: int, newId: int)) { + assert t.oldId == t.newId, + format("transparent reconnect changed session id {0} -> {1}", t.oldId, t.newId); + } + } +} + +spec CleanShutdown observes eSpecShutdownStarted, eSpecClosed { + var serverClosed: bool; + var clientClosed: bool; + + start cold state BeforeShutdown { + on eSpecClosed do (atServer: bool) { note(atServer); } + on eSpecShutdownStarted do { + if (serverClosed && clientClosed) { + goto AllClosed; + } else { + goto Waiting; + } + } + } + + hot state Waiting { + on eSpecClosed do (atServer: bool) { + note(atServer); + if (serverClosed && clientClosed) { + goto AllClosed; + } + } + ignore eSpecShutdownStarted; + } + + cold state AllClosed { + ignore eSpecClosed, eSpecShutdownStarted; + } + + fun note(atServer: bool) { + if (atServer) { + serverClosed = true; + } else { + clientClosed = true; + } + } +} diff --git a/verification/p/PSrc/Client.p b/verification/p/PSrc/Client.p new file mode 100644 index 00000000..8f4e3215 --- /dev/null +++ b/verification/p/PSrc/Client.p @@ -0,0 +1,457 @@ +/***************************************************************************** +ClientTransport: the client-side session state machine. + +State mapping to transport/sessionStateMachine/transitions.ts: + Idle — no session exists (TS: session deleted; a new call re-creates) + Connecting — SessionNoConnection + SessionBackingOff + SessionConnecting + collapsed: backoff duration is abstracted away (only orderings + matter, and each P state entry is its own scheduling point) + Handshaking — SessionHandshaking + Connected — SessionConnected + Dead — retry budget exhausted (conn_retry_exceeded) or fatal + handshake rejection: transport stays down, every call + resolves UNEXPECTED_DISCONNECT + Done — after eShutdown + +Session state carried across reconnects (TS inheritSharedSession): sessionId, +seqNum, seqSent, ackNum, sendBuffer (+ maxAckSeen, the model's explicit +"highest peer ack seen", implicit in TS). + +Grace period: armed (gen++) at session creation and on Connected->disconnected, +NOT re-armed across Connecting/Handshaking cycles (TS: absolute deadline), +cancelled (gen++ without re-arm) on entering Connected. +*****************************************************************************/ + +machine ClientTransport { + var orch: machine; + var server: machine; + var timer: machine; + var maxRetries: int; + var retriesLeft: int; + var corruptBudget: int; + + var hasSession: bool; + var sessionId: int; + var sessCounter: int; + var seqNum: int; // next seq to assign + var seqSent: int; // last seq written to a live connection (-1 = none) + var ackNum: int; // next expected inbound seq + var maxAckSeen: int; // highest peer ack processed + var sendBuffer: seq[tMsg]; + var pendingCalls: map[int, bool]; // issued, unresolved payloads + + var conn: machine; + var connId: int; // -1 = no current connection + var connCounter: int; + var graceGen: int; + var hadConnection: bool; // this session previously reached Connected + var credential: int; // current handshake metadata; construct() may refresh it + var streamWidth: int; + // per logical stream key: wire sid, requests nsent, half-closed by us, + // response pipe closed by the server + var streams: map[int, (sid: int, nsent: int, closedLocal: bool, serverClosed: bool)]; + + start state Boot { + entry (cfg: tClientCfg) { + orch = cfg.orch; + server = cfg.server; + maxRetries = cfg.maxRetries; + retriesLeft = cfg.maxRetries; + corruptBudget = cfg.corruptBudget; + streamWidth = cfg.streamWidth; + timer = new EchoTimer(this); + connId = -1; + credential = 1; + goto Idle; + } + } + + // No session. Everything that arrives here is stale except a new call. + state Idle { + on eAppSend do (p: int) { + newSession(); + enqueueApp(p); + goto Connecting; + } + on eShutdown do { + announce eSpecClosed, false; + goto Done; + } + on eConstructDone do (t: (sessGen: int, defers: int)) { handleConstructDone(t, false); } + ignore eConnClosed, eDeliverMsg, eDeliverHsResp, eConnEstablished, eGraceFired; + } + + state Connecting { + entry { + if (retriesLeft <= 0) { + destroySession(); + goto Dead; + } else { + retriesLeft = retriesLeft - 1; + connCounter = connCounter + 1; + connId = connCounter; + conn = new WsConnection((client = this, server = server, connId = connId)); + send orch, eConnCreated, conn; + } + } + on eConnEstablished do (e: (connId: int, conn: machine)) { + if (e.connId == connId) { + goto Handshaking; + } + } + on eConnClosed do (cid: int) { + if (cid == connId) { + connId = -1; + goto Connecting; + } + } + on eGraceFired do (g: int) { + if (g == graceGen) { + destroySession(); + goto Idle; + } + } + on eAppSend do (p: int) { enqueueApp(p); } + on eDeliverMsg do (w: tWireMsg) { assert w.connId != connId, "msg on conn before established"; } + on eDeliverHsResp do (h: tHsResp) { assert h.connId != connId, "hs resp before hs req"; } + on eConstructDone do (t: (sessGen: int, defers: int)) { handleConstructDone(t, false); } + on eShutdown do { shutdownNow(); goto Done; } + } + + state Handshaking { + entry { + send conn, eSendHsReq, (connId = connId, conn = conn, sessionId = sessionId, + nextExpectedSeq = ackNum, nextSentSeq = nextSeqLocal(), + isReconnect = hadConnection, meta = construct()); + } + on eDeliverHsResp do (h: tHsResp) { + if (h.connId != connId) { + return; // stale connection + } + if (!h.ok) { + send conn, eConnCloseCmd; + connId = -1; + if (h.retriable) { + // SESSION_STATE_MISMATCH: delete the session and reconnect with a + // FRESH session (seq/ack = 0). In-flight calls die. + destroySession(); + newSession(); + goto Connecting; + } else { + // fatal rejection: transport stays down + destroySession(); + goto Dead; + } + } else if (h.sessionId != sessionId) { + // TS: a mismatched session id in an ok response is itself fatal + send conn, eConnCloseCmd; + connId = -1; + destroySession(); + goto Dead; + } else { + goto Connected; + } + } + on eConnClosed do (cid: int) { + if (cid == connId) { + connId = -1; + goto Connecting; + } + } + on eGraceFired do (g: int) { + if (g == graceGen) { + destroySession(); + goto Idle; + } + } + on eAppSend do (p: int) { enqueueApp(p); } + on eDeliverMsg do (w: tWireMsg) { assert w.connId != connId, "msg before handshake resp"; } + on eConstructDone do (t: (sessGen: int, defers: int)) { handleConstructDone(t, false); } + ignore eConnEstablished; + on eShutdown do { shutdownNow(); goto Done; } + } + + state Connected { + entry { + hadConnection = true; + graceGen = graceGen + 1; // cancel grace (no re-arm) + retriesLeft = maxRetries; // TS restores the retry budget on connect + replayBuffer(); + maybeCorrupt(); + } + on eDeliverMsg do (w: tWireMsg) { + if (w.connId != connId) { + return; // stale connection + } + if (w.msg.seqn < ackNum) { + return; // duplicate: silently discarded + } + // seq > ack from the (honest) server would be a protocol invariant + // violation — the TS client logs `invariant-violation` and closes the + // connection. In this model the server is always honest, so it is + // simply unreachable. + assert w.msg.seqn == ackNum, + format("client received out-of-order seq {0}, expected {1}", w.msg.seqn, ackNum); + acceptMsg(w.msg); + } + on eConnClosed do (cid: int) { + if (cid == connId) { + connId = -1; + graceGen = graceGen + 1; + send timer, eStartGrace, graceGen; + goto Connecting; + } + } + on eAppSend do (p: int) { + transmit(enqueueApp(p)); + maybeCorrupt(); + } + on eDeliverHsResp do (h: tHsResp) { assert h.connId != connId, "duplicate handshake resp"; } + on eConstructDone do (t: (sessGen: int, defers: int)) { handleConstructDone(t, true); } + ignore eConnEstablished, eGraceFired; + on eShutdown do { shutdownNow(); goto Done; } + } + + state Dead { + on eAppSend do (p: int) { + announce eSpecResolved, (payload = p, ok = false); + send orch, eCallResolved, (payload = p, ok = false); + } + on eShutdown do { + announce eSpecClosed, false; + goto Done; + } + ignore eConnClosed, eDeliverMsg, eDeliverHsResp, eConnEstablished, eGraceFired, + eConstructDone; + } + + state Done { + ignore eAppSend, eShutdown, eConnClosed, eDeliverMsg, eDeliverHsResp, + eConnEstablished, eGraceFired, eConstructDone; + } + + /********************************* session **********************************/ + + fun newSession() { + sessCounter = sessCounter + 1; + sessionId = sessCounter; + hasSession = true; + seqNum = 0; + seqSent = -1; + ackNum = 0; + maxAckSeen = 0; + sendBuffer = default(seq[tMsg]); + pendingCalls = default(map[int, bool]); + hadConnection = false; + streams = default(map[int, (sid: int, nsent: int, closedLocal: bool, serverClosed: bool)]); + // TS arms the grace period at session creation (createUnconnectedSession) + graceGen = graceGen + 1; + send timer, eStartGrace, graceGen; + } + + fun destroySession() { + resolveAllDisconnect(); + hasSession = false; + announce eSpecReset; + if (connId != -1) { + send conn, eConnCloseCmd; + connId = -1; + } + } + + fun shutdownNow() { + if (hasSession) { + destroySession(); + } + announce eSpecClosed, false; + } + + fun resolveAllDisconnect() { + var ks: seq[int]; + var i: int; + ks = keys(pendingCalls); + i = 0; + while (i < sizeof(ks)) { + announce eSpecResolved, (payload = ks[i], ok = false); + send orch, eCallResolved, (payload = ks[i], ok = false); + i = i + 1; + } + pendingCalls = default(map[int, bool]); + } + + /********************************** wire ***********************************/ + + fun nextSeqLocal(): int { + if (sizeof(sendBuffer) > 0) { + return sendBuffer[0].seqn; + } + return seqNum; + } + + // Mirrors IdentifiedSession.send: stamp seq/ack once, buffer, bump seq. + fun enqueueMsg(kind: tMsgKind, p: int, sid: int, sopen: bool, sclose: bool): tMsg { + var m: tMsg; + m = (seqn = seqNum, ack = ackNum, kind = kind, payload = p, + sid = sid, sopen = sopen, sclose = sclose); + seqNum = seqNum + 1; + sendBuffer += (sizeof(sendBuffer), m); + assertBufferWindow(sendBuffer, seqNum, maxAckSeen); + return m; + } + + // Route an app payload onto its stream: OPEN on the first message of a + // stream instance, half-CLOSE on the last. Stream instances are per + // session (a session reset kills every stream; later payloads open fresh + // streams on the replacement session). + fun enqueueApp(p: int): tMsg { + var k: int; + var st: (sid: int, nsent: int, closedLocal: bool, serverClosed: bool); + var isOpen: bool; + var isClose: bool; + pendingCalls[p] = true; + k = (p + streamWidth - 1) / streamWidth; + if (!(k in streams)) { + streams[k] = (sid = sessCounter * 100 + k, nsent = 0, + closedLocal = false, serverClosed = false); + isOpen = true; + } + st = streams[k]; + assert !st.closedLocal, "client wrote to a stream after closing its writer"; + st.nsent = st.nsent + 1; + if (st.nsent == streamWidth) { + isClose = true; // half-close: our writer closes, reader stays open + st.closedLocal = true; + } + streams[k] = st; + return enqueueMsg(MSG_APP, p, st.sid, isOpen, isClose); + } + + // C4-style inline invariant: never put an out-of-order seq on the wire. + fun transmit(m: tMsg) { + assert m.seqn <= seqSent + 1, + format("client would send out-of-order seq {0}, seqSent {1}", m.seqn, seqSent); + send conn, eSendMsg, (toServer = true, msg = m); + seqSent = m.seqn; + } + + // Mirrors SessionConnected.sendBufferedMessages: replay everything in order. + fun replayBuffer() { + var i: int; + i = 0; + while (i < sizeof(sendBuffer)) { + transmit(sendBuffer[i]); + i = i + 1; + } + } + + // Mirrors SessionConnected.updateBookkeeping. + fun acceptMsg(m: tMsg) { + ackNum = m.seqn + 1; + if (m.ack > maxAckSeen) { + maxAckSeen = m.ack; + } + while (sizeof(sendBuffer) > 0 && sendBuffer[0].seqn < m.ack) { + sendBuffer -= (0); + } + assertBufferWindow(sendBuffer, seqNum, maxAckSeen); + if (m.kind == MSG_APP) { + trackResponseStream(m); + announce eSpecDeliver, (atServer = false, payload = m.payload); + if (m.payload in pendingCalls) { + pendingCalls -= (m.payload); + announce eSpecResolved, (payload = m.payload, ok = true); + send orch, eCallResolved, (payload = m.payload, ok = true); + } + } else if (m.kind == MSG_ACK) { + // TS: a passive (non-heartbeating) side echoes an inbound heartbeat + transmit(enqueueMsg(MSG_ACK, 0, 0, false, false)); + } else if (m.kind == MSG_RH_REQ) { + // re-run metadata construction asynchronously (the TS construct() gap): + // the response send is bound to the session captured HERE + send this, eConstructDone, (sessGen = sessCounter, defers = 2); + } else { + assert false, "client received a MSG_RH_RESP"; + } + } + + // Half-close discipline on the response pipe: the server may write on a + // stream WE half-closed (that is the point of half-close), but nothing may + // arrive after the SERVER closed its writer. + fun trackResponseStream(m: tMsg) { + var k: int; + var st: (sid: int, nsent: int, closedLocal: bool, serverClosed: bool); + var ks: seq[int]; + var i: int; + ks = keys(streams); + i = 0; + while (i < sizeof(ks)) { + k = ks[i]; + st = streams[k]; + if (st.sid == m.sid) { + assert !st.serverClosed, + format("server nsent data on stream {0} after closing its writer", m.sid); + if (m.sclose) { + st.serverClosed = true; + streams[k] = st; + } + return; + } + i = i + 1; + } + // response for a stream this session no longer knows: impossible, since + // stream records live exactly as long as the session + assert false, format("client received a response for unknown stream {0}", m.sid); + } + + // TS handshakeExtensions.construct(): re-reads the (possibly refreshed) + // credential. The nondeterministic bump models a token refresh. + fun construct(): int { + if ($) { + credential = credential + 1; + } + return credential; + } + + // Completion of the async construct() for a rehandshake response. If the + // session it was bound to is gone, the bound send throws in TS — here we + // silently drop, which is the OBSERVABLE fixed behavior (stale metadata + // must not reach the replacement session). + fun handleConstructDone(t: (sessGen: int, defers: int), canTransmit: bool) { + var m: tMsg; + if (t.defers > 0 && $) { + send this, eConstructDone, (sessGen = t.sessGen, defers = t.defers - 1); + return; + } + if (!hasSession || t.sessGen != sessCounter) { + return; // session replaced during construct(): bound send throws + } + m = enqueueMsg(MSG_RH_RESP, construct(), 0, false, false); + if (canTransmit) { + transmit(m); + } + } + + /******************************** byzantine ********************************/ + + // T6: a misbehaving client emits extra bogus frames outside its own + // bookkeeping. A duplicate (seq < server.ack, always) must be silently + // dropped; a future seq (> server.ack, always) must make the server close + // the CONNECTION while the session survives. + fun maybeCorrupt() { + var m: tMsg; + if (corruptBudget <= 0 || connId == -1) { + return; + } + if ($) { + corruptBudget = corruptBudget - 1; + if (maxAckSeen > 0 && $) { + m = (seqn = maxAckSeen - 1, ack = ackNum, kind = MSG_APP, payload = 0, + sid = 0, sopen = false, sclose = false); + } else { + m = (seqn = seqNum + 1, ack = ackNum, kind = MSG_APP, payload = 0, + sid = 0, sopen = false, sclose = false); + } + send conn, eSendMsg, (toServer = true, msg = m); + } + } +} diff --git a/verification/p/PSrc/Events.p b/verification/p/PSrc/Events.p new file mode 100644 index 00000000..d6368fbb --- /dev/null +++ b/verification/p/PSrc/Events.p @@ -0,0 +1,194 @@ +/***************************************************************************** +River protocol model: shared types, events, and helper functions. + +The model covers the transport/session layer of PROTOCOL.md: seq/ack +exactly-once bookkeeping, the send buffer, the 4-case handshake, transparent +vs hard reconnects, and the session grace period. Payloads are monotonically +increasing ints issued by the test driver; the server echoes every accepted +app message back, which stands in for an rpc response (bidirectional traffic +exercises both directions' seq/ack windows and gives calls a natural +resolution point). +*****************************************************************************/ + +enum tMsgKind { + MSG_APP, + MSG_ACK, // heartbeat (AckBit); rides the normal seq/ack path + MSG_RH_REQ, // ControlRehandshakeRequest (server -> client), reserved streamId + MSG_RH_RESP // ControlRehandshakeResponse (client -> server); payload = credential +} + +enum tDropMode { + DROP_BOTH, // close event delivered to both sides + DROP_CLIENT_ONLY, // only the client learns the connection died + DROP_SERVER_ONLY, // only the server learns (needs watchdog; phase 2) + DROP_SILENT // neither side learns (phantom disconnect; phase 2) +} + +// A transport message. In the TS implementation the encoded bytes are cached +// in the send buffer, so retransmits are byte-identical: a replayed message +// carries its ORIGINAL (possibly stale) ack. tMsg mirrors that: it is stamped +// once at enqueue time and never re-stamped. +// +// Stream overlay (B-series): app messages belong to a stream. `sid` is the +// wire stream id (unique per stream instance, client-generated), `sopen` is +// the StreamOpenBit (first message of a stream, client-only), `sclose` is the +// StreamClosedBit (the sender closes its writer; the other side may keep +// writing — half-close). Control messages use sid 0 and no flags. +type tMsg = (seqn: int, ack: int, kind: tMsgKind, payload: int, + sid: int, sopen: bool, sclose: bool); + +// Messages delivered by a connection are tagged with the connection's id so +// endpoints can drop deliveries from a connection they no longer own +// (the TS impl detaches the old connection's listeners instead). +type tWireMsg = (connId: int, msg: tMsg); + +// Handshake frames are out-of-band (seq 0 / ack 0 in TS) and do not consume +// seq numbers. `conn` lets the server respond on the requesting connection. +type tHsReq = (connId: int, conn: machine, sessionId: int, + nextExpectedSeq: int, // = client session.ack + nextSentSeq: int, // = client session.nextSeq() + isReconnect: bool, // client session was previously Connected + meta: int); // handshake metadata (credential) +type tHsResp = (connId: int, ok: bool, sessionId: int, retriable: bool); + +type tConnCfg = (client: machine, server: machine, connId: int); +// streamWidth: app payloads per stream — payloads are assigned round-robin in +// blocks (width 2: payloads 1,2 -> stream 1; 3,4 -> stream 2; ...). The last +// payload of a block carries the client's half-close. +type tClientCfg = (orch: machine, server: machine, maxRetries: int, corruptBudget: int, + streamWidth: int); +// consumedGuardFixed models commit d7c0ec9: when true, the rehandshake +// teardown checks `session._isConsumed` first and returns silently on a stale +// handle; when false (the pre-fix code), touching a consumed session state +// crashes (an unhandled rejection in TS, an assertion here). +// zeroStateGuardFixed: when true, the server honors the handshake's +// isReconnect flag and rejects a reconnect to a session it lost even when +// the client's seq counters are all-zero; when false (the pre-fix protocol), +// such a reconnect is accepted as a NEW session and the client's replay +// re-delivers already-delivered payloads (the model checker found this; +// reproduced against the TS implementation in __tests__/zerostate.test.ts). +type tServerCfg = (orch: machine, expectHonest: bool, consumedGuardFixed: bool, + zeroStateGuardFixed: bool); +// dropMix selects which tDropMode values the fault injector may choose: +// 0 = DROP_BOTH only +// 1 = DROP_BOTH | DROP_CLIENT_ONLY +// 2 = any, including DROP_SERVER_ONLY and DROP_SILENT (phantom disconnect; +// requires the watchdog to unstick the unnotified side) +type tOrchCfg = (n: int, dropBudget: int, dropMix: int, restartBudget: int, + heartbeatBudget: int, rehandshakeBudget: int, corruptBudget: int, + expectHonest: bool, consumedGuardFixed: bool, + zeroStateGuardFixed: bool); + +/******************************* app-facing *********************************/ +event eAppSend: int; // driver -> client: issue call +event eCallResolved: (payload: int, ok: bool); // client -> driver +event eShutdown; // driver -> both endpoints +event eRestart; // driver -> server: lose all state +event eConnCreated: machine; // client -> driver (fault hook) + +/*************************** connection lifecycle ****************************/ +event eConnEstablished: (connId: int, conn: machine); // conn -> client +event eConnClosed: int; // conn -> endpoint (connId) +event eConnCloseCmd; // endpoint -> conn +event eFaultDrop: tDropMode; // driver -> conn +// Watchdog detection (C9b, untimed): a side that was NOT notified of a dead +// connection eventually notices (TS: lastInboundAt falls behind +// heartbeatsUntilDead * heartbeatIntervalMs and the watchdog closes the +// connection). Modeled as a deferred eConnClosed delivered by the dead +// connection to the unnotified side(s). The timing bound itself (C9a: a live +// peer is never falsely killed) is a wall-clock property outside an untimed +// model's reach. +event eWatchdogDetect: int; // conn -> itself (defers) +event eHeartbeatNudge; // driver -> server: fire one heartbeat +event eRehandshakeNudge; // driver -> server: refresh credentials + +/****************************** rehandshake *********************************/ +// The TS implementation has two async gaps in the rehandshake exchange, both +// implicated in the d7c0ec9 bug class: +// - the client's construct() (rebuilding handshake metadata) — a hard +// reconnect during it must make the bound send throw, not deliver stale +// metadata to the replacement session +// - the server's validate() — it can complete after the session state it +// captured was consumed by a transition +// Both are modeled as deferred self-events carrying the identity captured at +// the start of the gap: sessGen (which session) and connEpoch (which +// Connected-state instance — TS's linear-typed, consumable state object). +event eConstructDone: (sessGen: int, defers: int); // client -> itself +event eValidateDone: (sessGen: int, connEpoch: int, reject: bool, + meta: int, defers: int); // server -> itself + +/******************************** wire traffic ******************************/ +event eSendMsg: (toServer: bool, msg: tMsg); // endpoint -> conn +event eDeliverMsg: tWireMsg; // conn -> endpoint +event eSendHsReq: tHsReq; // client -> conn +event eDeliverHsReq: tHsReq; // conn -> server +event eSendHsResp: tHsResp; // server -> conn +event eDeliverHsResp: tHsResp; // conn -> client + +/********************************** timers **********************************/ +// Timers are modeled as generation-guarded fire events routed through a +// separate EchoTimer machine: the fire lands in the owner's queue at a +// scheduler-chosen point, so the checker explores every firing interleaving. +// A generation mismatch means the timer was cancelled (owner bumped its gen). +// +// A naive immediate echo would land the fire EARLY in the owner's FIFO queue +// in most schedules (before the handshake even completes), biasing the search +// toward grace-killed sessions. The timer therefore defers the fire a +// nondeterministic, budgeted number of times via self-sends — each defer +// pushes the fire to the back of the owner's arrival order, so late firings +// (the realistic case: grace is seconds, a reconnect is milliseconds) are +// explored too. +event eStartGrace: int; // owner -> timer (generation) +event eGraceFired: int; // timer -> owner (generation) +event eTimerDefer: (gen: int, defers: int); // timer -> itself + +/***************************** spec announcements ****************************/ +event eSpecCall: int; // driver issued a call +event eSpecResolved: (payload: int, ok: bool); // call resolved (once) +event eSpecDeliver: (atServer: bool, payload: int); // app payload accepted +event eSpecReset; // a session was destroyed (hard reset) +event eSpecTransparentReconnect: (oldId: int, newId: int); +event eSpecShutdownStarted; +event eSpecClosed: bool; // endpoint closed (atServer) + +/********************************* helpers **********************************/ + +// C3 (strengthened): the send buffer holds exactly the window +// [maxAckSeen, seqNum) of contiguous seqs — an unacked message is never +// dropped and an acked one never lingers. Mirrors the TS invariant that the +// hegel tests check via the `invariant-violation` log oracle. +fun assertBufferWindow(buffer: seq[tMsg], seqNum: int, maxAckSeen: int) { + var i: int; + if (sizeof(buffer) == 0) { + assert maxAckSeen == seqNum, + format("send buffer empty but window [{0}, {1}) not empty: unacked messages were dropped", maxAckSeen, seqNum); + } else { + assert buffer[0].seqn == maxAckSeen, + format("send buffer starts at {0}, expected peer-acked lower bound {1}", buffer[0].seqn, maxAckSeen); + i = 0; + while (i < sizeof(buffer)) { + assert buffer[i].seqn == buffer[0].seqn + i, + format("send buffer not contiguous at index {0}", i); + i = i + 1; + } + assert buffer[sizeof(buffer) - 1].seqn + 1 == seqNum, + format("send buffer ends at {0}, expected seqNum {1}", buffer[sizeof(buffer) - 1].seqn + 1, seqNum); + } +} + +machine EchoTimer { + var target: machine; + start state Idle { + entry (t: machine) { target = t; } + on eStartGrace do (gen: int) { + send this, eTimerDefer, (gen = gen, defers = 3); + } + on eTimerDefer do (t: (gen: int, defers: int)) { + if (t.defers > 0 && $) { + send this, eTimerDefer, (gen = t.gen, defers = t.defers - 1); + } else { + send target, eGraceFired, t.gen; + } + } + } +} diff --git a/verification/p/PSrc/Network.p b/verification/p/PSrc/Network.p new file mode 100644 index 00000000..fa11b749 --- /dev/null +++ b/verification/p/PSrc/Network.p @@ -0,0 +1,91 @@ +/***************************************************************************** +WsConnection: one machine per WebSocket connection instance. + +P guarantees FIFO delivery per sender->receiver machine pair, so routing all +traffic through this machine gives WebSocket's ordered reliable delivery +within a connection for free, while messages traveling through DIFFERENT +connection instances race naturally — exactly the "old socket's buffered +frames arrive after the new handshake" interleaving the d7c0ec9 bug lived in. + +Faults: eFaultDrop kills the connection and notifies both sides, one side, or +neither (phantom disconnect), per tDropMode. eConnCloseCmd is an +endpoint-initiated close (TS conn.close()): both sides get the close event. +Once Closed, all in-transit sends die on the wire. +*****************************************************************************/ + +machine WsConnection { + var client: machine; + var server: machine; + var connId: int; + var dropDefers: int; + var clientUnnotified: bool; + var serverUnnotified: bool; + + start state Open { + entry (cfg: tConnCfg) { + client = cfg.client; + server = cfg.server; + connId = cfg.connId; + dropDefers = 4; + send client, eConnEstablished, (connId = connId, conn = this); + } + on eSendMsg do (s: (toServer: bool, msg: tMsg)) { + if (s.toServer) { + send server, eDeliverMsg, (connId = connId, msg = s.msg); + } else { + send client, eDeliverMsg, (connId = connId, msg = s.msg); + } + } + on eSendHsReq do (h: tHsReq) { send server, eDeliverHsReq, h; } + on eSendHsResp do (h: tHsResp) { send client, eDeliverHsResp, h; } + // The fault injector decides to drop a connection when it is created; the + // budgeted nondeterministic defers below move the actual drop point to an + // arbitrary later position in the connection's message stream (the same + // shape as the hegel write-schedule generator's `null` fault points). + on eFaultDrop do (mode: tDropMode) { + if (dropDefers > 0 && $) { + dropDefers = dropDefers - 1; + send this, eFaultDrop, mode; + } else { + if (mode == DROP_BOTH || mode == DROP_CLIENT_ONLY) { + send client, eConnClosed, connId; + } else { + clientUnnotified = true; + } + if (mode == DROP_BOTH || mode == DROP_SERVER_ONLY) { + send server, eConnClosed, connId; + } else { + serverUnnotified = true; + } + if (clientUnnotified || serverUnnotified) { + // the unnotified side's heartbeat watchdog eventually notices + send this, eWatchdogDetect, 3; + } + goto Closed; + } + } + on eConnCloseCmd do { + send client, eConnClosed, connId; + send server, eConnClosed, connId; + goto Closed; + } + } + + state Closed { + on eWatchdogDetect do (defers: int) { + if (defers > 0 && $) { + send this, eWatchdogDetect, defers - 1; + } else { + if (clientUnnotified) { + send client, eConnClosed, connId; + clientUnnotified = false; + } + if (serverUnnotified) { + send server, eConnClosed, connId; + serverUnnotified = false; + } + } + } + ignore eSendMsg, eSendHsReq, eSendHsResp, eFaultDrop, eConnCloseCmd; + } +} diff --git a/verification/p/PSrc/Server.p b/verification/p/PSrc/Server.p new file mode 100644 index 00000000..92c9a811 --- /dev/null +++ b/verification/p/PSrc/Server.p @@ -0,0 +1,351 @@ +/***************************************************************************** +ServerTransport: the server-side session state machine. + +The server never dials. It holds at most one session (single-client model) +in a data-driven `Running` state (the TS server juggles a live session plus +concurrent pending handshakes, so flat states don't fit). A connection in +TS's SessionWaitingForHandshake is implicit here: a WsConnection whose +handshake request hasn't been processed yet. + +onHandshakeRequest implements the four connect cases of PROTOCOL.md / +transport/server.ts: + 1. transparent reconnect — old session with the same id; reject + SESSION_STATE_MISMATCH if the client or server is "in the future", + else demote the old connection and adopt the new one with inherited + seq/ack/sendBuffer + 2. hard reconnect — old session with a DIFFERENT id: delete it, + fall through + 3. unknown session — no session but nonzero expected state: reject + SESSION_STATE_MISMATCH (retriable) + 4. new session — adopt the client-supplied session id, zero state + +eRestart models a server that lost all state (forces case 2/3 on reconnect). +*****************************************************************************/ + +machine ServerTransport { + var orch: machine; + var timer: machine; + var expectHonest: bool; // false in byzantine scenarios (T6) + + var hasSession: bool; + var sessId: int; + var seqNum: int; + var seqSent: int; + var ackNum: int; + var maxAckSeen: int; + var sendBuffer: seq[tMsg]; + + var conn: machine; + var connId: int; // -1 = session disconnected (grace running) + var graceGen: int; + + var meta: int; // stored handshake metadata (credential) + var sessGen: int; // session identity: bumped on destroy/create + var connEpoch: int; // Connected-state instance identity: bumped on + // every adopt/demote/close — models the TS + // linear-typed (consumable) session state object + var rhOutstanding: bool; // one rehandshake round in flight at a time + var consumedGuardFixed: bool; + var zeroStateGuardFixed: bool; + // stream overlay: per wire stream id, opened/half-closed-by-client state. + // Lives exactly as long as the session (cleared on destroy/create). + var streams: map[int, (clientClosed: bool, echoed: int)]; + + start state Boot { + entry (cfg: tServerCfg) { + orch = cfg.orch; + expectHonest = cfg.expectHonest; + consumedGuardFixed = cfg.consumedGuardFixed; + zeroStateGuardFixed = cfg.zeroStateGuardFixed; + timer = new EchoTimer(this); + connId = -1; + goto Running; + } + } + + state Running { + on eDeliverHsReq do (h: tHsReq) { handleHandshake(h); } + on eDeliverMsg do (w: tWireMsg) { handleMsg(w); } + on eConnClosed do (cid: int) { + if (hasSession && cid == connId) { + connId = -1; + connEpoch = connEpoch + 1; // ConnectedToNoConnection consumes the state + graceGen = graceGen + 1; + send timer, eStartGrace, graceGen; + } + } + on eGraceFired do (g: int) { + if (hasSession && connId == -1 && g == graceGen) { + destroySession(); + } + } + on eRestart do { + if (hasSession) { + destroySession(); + } + } + // Only the server actively heartbeats (TS startActiveHeartbeat). The + // heartbeat is an AckBit control message through the NORMAL send path: + // it consumes a seq and occupies the send buffer, exactly as in TS. + on eHeartbeatNudge do { + if (hasSession && connId != -1) { + sendNow(MSG_ACK, 0); + } + } + // Server-initiated credential refresh (TS scheduleRehandshake / + // requestRehandshakeNow): a ControlRehandshakeRequest through the normal + // seq/ack send path. + on eRehandshakeNudge do { + if (hasSession && connId != -1 && !rhOutstanding) { + rhOutstanding = true; + sendNow(MSG_RH_REQ, 0); + } + } + on eValidateDone do (v: (sessGen: int, connEpoch: int, reject: bool, meta: int, defers: int)) { + handleValidateDone(v); + } + on eShutdown do { + if (hasSession) { + destroySession(); + } + announce eSpecClosed, true; + goto Done; + } + } + + state Done { + ignore eDeliverHsReq, eDeliverMsg, eConnClosed, eGraceFired, eRestart, eShutdown, + eHeartbeatNudge, eRehandshakeNudge, eValidateDone; + } + + /******************************** handshake ********************************/ + + fun handleHandshake(h: tHsReq) { + if (hasSession && sessId == h.sessionId) { + // case 1: transparent reconnect to the existing session + if (h.nextSentSeq > ackNum) { + // client is in the future + rejectRetriable(h); + return; + } + if (nextSeqLocal() > h.nextExpectedSeq) { + // server is in the future + rejectRetriable(h); + return; + } + if (connId != -1) { + // demote the old live connection before adopting the new one + send conn, eConnCloseCmd; + } + announce eSpecTransparentReconnect, (oldId = sessId, newId = h.sessionId); + + adoptConn(h); + return; + } + if (hasSession) { + // case 2: hard reconnect — the client wants a session we don't have + destroySession(); + } + if (h.nextSentSeq > 0 || h.nextExpectedSeq > 0 + || (zeroStateGuardFixed && h.isReconnect)) { + // case 3: reconnect to an unknown session — nothing to salvage. + // The isReconnect check closes the zero-state window: without it, a + // reconnecting client that never received an ack is indistinguishable + // from a new session and its replay re-delivers to handlers. + rejectRetriable(h); + return; + } + // case 4: new session, adopt the client-supplied id with zero state + hasSession = true; + sessGen = sessGen + 1; + rhOutstanding = false; + streams = default(map[int, (clientClosed: bool, echoed: int)]); + sessId = h.sessionId; + seqNum = 0; + seqSent = -1; + ackNum = 0; + maxAckSeen = 0; + sendBuffer = default(seq[tMsg]); + adoptConn(h); + } + + fun adoptConn(h: tHsReq) { + if (connId != -1) { + connEpoch = connEpoch + 1; // demotion consumed the old Connected state + } + connId = h.connId; + conn = h.conn; + connEpoch = connEpoch + 1; // a fresh Connected-state instance + meta = h.meta; // storeSessionMetadata (re-validated on every handshake) + graceGen = graceGen + 1; // cancel grace + // Response first, then the buffer replay: FIFO through the connection + // guarantees the client sees them in that order. + send conn, eSendHsResp, (connId = h.connId, ok = true, sessionId = sessId, retriable = false); + replayBuffer(); + } + + fun rejectRetriable(h: tHsReq) { + send h.conn, eSendHsResp, (connId = h.connId, ok = false, sessionId = h.sessionId, retriable = true); + // TS deletePendingSession closes the rejected connection + send h.conn, eConnCloseCmd; + } + + fun destroySession() { + hasSession = false; + sessGen = sessGen + 1; + connEpoch = connEpoch + 1; + rhOutstanding = false; + streams = default(map[int, (clientClosed: bool, echoed: int)]); + announce eSpecReset; + if (connId != -1) { + send conn, eConnCloseCmd; + connId = -1; + } + } + + /********************************** wire ***********************************/ + + fun nextSeqLocal(): int { + if (sizeof(sendBuffer) > 0) { + return sendBuffer[0].seqn; + } + return seqNum; + } + + fun handleMsg(w: tWireMsg) { + if (!hasSession || w.connId != connId) { + return; // stale connection or no session + } + if (w.msg.seqn < ackNum) { + return; // duplicate: silently discarded + } + if (w.msg.seqn > ackNum) { + // TS logs `invariant-violation` and closes the CONNECTION to recover by + // re-handshake with the session intact. Among honest peers this is + // unreachable — asserting that is the model's C4 oracle. + assert !expectHonest, + format("server received out-of-order seq {0}, expected {1}", w.msg.seqn, ackNum); + send conn, eConnCloseCmd; + return; + } + // accept: mirrors SessionConnected.updateBookkeeping + ackNum = w.msg.seqn + 1; + if (w.msg.ack > maxAckSeen) { + maxAckSeen = w.msg.ack; + } + while (sizeof(sendBuffer) > 0 && sendBuffer[0].seqn < w.msg.ack) { + sendBuffer -= (0); + } + assertBufferWindow(sendBuffer, seqNum, maxAckSeen); + if (w.msg.kind == MSG_APP) { + trackRequestStream(w.msg); + announce eSpecDeliver, (atServer = true, payload = w.msg.payload); + // echo = the response; when the client half-closed and this is the last + // echo, the server closes its writer too (full close, upload-style) + sendNow2(MSG_APP, w.msg.payload, w.msg.sid, streams[w.msg.sid].clientClosed); + } else if (w.msg.kind == MSG_RH_RESP) { + // TS onRehandshakeResponse: the response arrived (deadline cleared), + // then `await validate(...)` — an async gap during which the session + // state captured here can be consumed by a transition. The reject + // verdict models a custom validator rejecting the refreshed metadata. + if (rhOutstanding) { + rhOutstanding = false; + send this, eValidateDone, (sessGen = sessGen, connEpoch = connEpoch, + reject = $, meta = w.msg.payload, defers = 2); + } + } + // an inbound MSG_ACK updates bookkeeping only; the active heartbeater + // does not echo heartbeats back + } + + // Completion of the async validate(). On success, TS stores the metadata + // only if the session it validated against is still the live one. On + // failure it calls teardownForFailedRehandshake — where d7c0ec9 lives: + // pre-fix: `if (this.sessions.get(session.to) !== session) return;` + // reads `session.to` on a possibly-consumed state proxy, which + // THROWS (an unhandled rejection at runtime) + // post-fix: `if (session._isConsumed) return;` guards it first + fun handleValidateDone(v: (sessGen: int, connEpoch: int, reject: bool, meta: int, defers: int)) { + if (v.defers > 0 && $) { + send this, eValidateDone, (sessGen = v.sessGen, connEpoch = v.connEpoch, + reject = v.reject, meta = v.meta, defers = v.defers - 1); + return; + } + if (!v.reject) { + // only store if it's still the session (and state instance) we + // validated against — don't clobber fresher metadata + if (v.sessGen == sessGen && v.connEpoch == connEpoch) { + meta = v.meta; + } + return; + } + // failed rehandshake: tear the session down + if (consumedGuardFixed) { + if (v.sessGen != sessGen || v.connEpoch != connEpoch) { + return; // d7c0ec9 guard: stale handle, cleanup belongs elsewhere + } + } else { + assert v.sessGen == sessGen && v.connEpoch == connEpoch, + "d7c0ec9: rehandshake teardown accessed a consumed session state (unhandled rejection in TS)"; + } + if (hasSession) { + destroySession(); + } + } + + // B-series flag discipline on the request pipe: a stream MUST be opened by + // its first message and only by its first message, and the client MUST NOT + // write after closing its writer. (The seq layer dedups replays before this + // runs, so a replayed OPEN after a transparent reconnect never reaches here + // twice — but a bookkeeping bug that broke dedup would trip these.) + fun trackRequestStream(m: tMsg) { + var st: (clientClosed: bool, echoed: int); + if (m.sopen) { + assert !(m.sid in streams), format("stream {0} opened twice", m.sid); + streams[m.sid] = (clientClosed = false, echoed = 0); + } else { + assert m.sid in streams, format("data on stream {0} before open", m.sid); + assert !streams[m.sid].clientClosed, + format("client data on stream {0} after its half-close", m.sid); + } + if (m.sclose) { + st = streams[m.sid]; + st.clientClosed = true; + streams[m.sid] = st; + } + } + + // The normal send path (consumes a seq, enters the send buffer, replayed on + // reconnect). MSG_APP = the "rpc response" echo; MSG_ACK = a heartbeat. + fun sendNow(kind: tMsgKind, p: int) { + sendNow2(kind, p, 0, false); + } + + fun sendNow2(kind: tMsgKind, p: int, sid: int, sclose: bool) { + var m: tMsg; + m = (seqn = seqNum, ack = ackNum, kind = kind, payload = p, + sid = sid, sopen = false, sclose = sclose); + seqNum = seqNum + 1; + sendBuffer += (sizeof(sendBuffer), m); + assertBufferWindow(sendBuffer, seqNum, maxAckSeen); + if (connId != -1) { + transmit(m); + } + } + + fun transmit(m: tMsg) { + assert m.seqn <= seqSent + 1, + format("server would send out-of-order seq {0}, seqSent {1}", m.seqn, seqSent); + send conn, eSendMsg, (toServer = false, msg = m); + seqSent = m.seqn; + } + + fun replayBuffer() { + var i: int; + i = 0; + while (i < sizeof(sendBuffer)) { + transmit(sendBuffer[i]); + i = i + 1; + } + } +} diff --git a/verification/p/PTst/Testscript.p b/verification/p/PTst/Testscript.p new file mode 100644 index 00000000..51822106 --- /dev/null +++ b/verification/p/PTst/Testscript.p @@ -0,0 +1,269 @@ +/***************************************************************************** +Test scenarios. + +The Orchestrator is the test driver: it spawns the server and client, issues +N calls, injects a budgeted number of nondeterministic faults, and shuts +everything down once every call has resolved. A fault decision is made when a +connection is created; the eFaultDrop event's DELIVERY point is chosen by the +scheduler, so the drop lands at an arbitrary position in that connection's +message stream (the same shape as the hegel write-schedule generator). + +The session grace period races with reconnects in every scenario for free: +grace firing is a scheduler choice, not a scripted step. + + tcHappyPath — no faults: pure seq/ack + handshake sanity + tcDrops — up to 3 connection drops (both-sided or client-only): + the core transparent-reconnect test (C1/C2/C3/C6) + tcRestart — a server restart plus up to 1 drop: hard reconnect, + UNEXPECTED_DISCONNECT resolution (C5) + tcByzantine — a misbehaving client injects duplicate/future seqs: the + server must drop duplicates silently and close the + connection (not the session) on future seqs + tcPhantom — one-sided and silent drops (phantom disconnects): the + unnotified side is unstuck only by the heartbeat watchdog's + eventual detection (C9b, untimed); heartbeats ride the + normal seq/ack path + tcRehandshake — server-initiated credential refreshes racing connection + drops; the d7c0ec9 consumed-handle guard is ON (expect no + bug) + tcD7c0ec9Regression + — same scenario with the PRE-FIX teardown (no consumed-handle + guard): the checker must FIND the crash — a model-level + regression test for the d7c0ec9 bug class +*****************************************************************************/ + +machine Orchestrator { + var n: int; + var dropsLeft: int; + var dropMix: int; + var restartsLeft: int; + var heartbeatsLeft: int; + var rehandshakesLeft: int; + var resolvedCount: int; + var client: machine; + var server: machine; + + start state Boot { + entry (cfg: tOrchCfg) { + var i: int; + n = cfg.n; + dropsLeft = cfg.dropBudget; + dropMix = cfg.dropMix; + restartsLeft = cfg.restartBudget; + heartbeatsLeft = cfg.heartbeatBudget; + rehandshakesLeft = cfg.rehandshakeBudget; + server = new ServerTransport((orch = this, expectHonest = cfg.expectHonest, + consumedGuardFixed = cfg.consumedGuardFixed, + zeroStateGuardFixed = cfg.zeroStateGuardFixed)); + client = new ClientTransport((orch = this, server = server, + maxRetries = 3, corruptBudget = cfg.corruptBudget, + streamWidth = 2)); + i = 1; + while (i <= n) { + announce eSpecCall, i; + send client, eAppSend, i; + i = i + 1; + } + goto Run; + } + } + + state Run { + on eConnCreated do (c: machine) { + maybeRestart(); + maybeHeartbeat(); + maybeRehandshake(); + if (dropsLeft > 0 && $) { + dropsLeft = dropsLeft - 1; + send c, eFaultDrop, pickDropMode(); + } + } + on eCallResolved do (r: (payload: int, ok: bool)) { + resolvedCount = resolvedCount + 1; + maybeRestart(); + maybeHeartbeat(); + maybeRehandshake(); + if (resolvedCount == n) { + announce eSpecShutdownStarted; + send client, eShutdown; + send server, eShutdown; + } + } + } + + fun pickDropMode(): tDropMode { + var pick: int; + if (dropMix == 0) { + return DROP_BOTH; + } + if (dropMix == 1) { + if ($) { + return DROP_CLIENT_ONLY; + } + return DROP_BOTH; + } + pick = choose(4); + if (pick == 0) { + return DROP_BOTH; + } + if (pick == 1) { + return DROP_CLIENT_ONLY; + } + if (pick == 2) { + return DROP_SERVER_ONLY; + } + return DROP_SILENT; + } + + fun maybeRestart() { + if (restartsLeft > 0 && $) { + restartsLeft = restartsLeft - 1; + send server, eRestart; + } + } + + fun maybeHeartbeat() { + if (heartbeatsLeft > 0 && $) { + heartbeatsLeft = heartbeatsLeft - 1; + send server, eHeartbeatNudge; + } + } + + fun maybeRehandshake() { + if (rehandshakesLeft > 0 && $) { + rehandshakesLeft = rehandshakesLeft - 1; + send server, eRehandshakeNudge; + } + } +} + +machine MainHappyPath { + start state Init { + entry { + new Orchestrator((n = 4, dropBudget = 0, dropMix = 0, restartBudget = 0, + heartbeatBudget = 1, rehandshakeBudget = 0, corruptBudget = 0, + expectHonest = true, consumedGuardFixed = true, + zeroStateGuardFixed = true)); + } + } +} + +machine MainDrops { + start state Init { + entry { + new Orchestrator((n = 4, dropBudget = 3, dropMix = 1, restartBudget = 0, + heartbeatBudget = 1, rehandshakeBudget = 0, corruptBudget = 0, + expectHonest = true, consumedGuardFixed = true, + zeroStateGuardFixed = true)); + } + } +} + +machine MainRestart { + start state Init { + entry { + new Orchestrator((n = 3, dropBudget = 1, dropMix = 1, restartBudget = 1, + heartbeatBudget = 1, rehandshakeBudget = 0, corruptBudget = 0, + expectHonest = true, consumedGuardFixed = true, + zeroStateGuardFixed = true)); + } + } +} + +machine MainByzantine { + start state Init { + entry { + new Orchestrator((n = 3, dropBudget = 1, dropMix = 0, restartBudget = 0, + heartbeatBudget = 0, rehandshakeBudget = 0, corruptBudget = 2, + expectHonest = false, consumedGuardFixed = true, + zeroStateGuardFixed = true)); + } + } +} + +machine MainPhantom { + start state Init { + entry { + new Orchestrator((n = 3, dropBudget = 2, dropMix = 2, restartBudget = 0, + heartbeatBudget = 2, rehandshakeBudget = 0, corruptBudget = 0, + expectHonest = true, consumedGuardFixed = true, + zeroStateGuardFixed = true)); + } + } +} + +machine MainRehandshake { + start state Init { + entry { + new Orchestrator((n = 3, dropBudget = 2, dropMix = 1, restartBudget = 0, + heartbeatBudget = 0, rehandshakeBudget = 2, corruptBudget = 0, + expectHonest = true, consumedGuardFixed = true, + zeroStateGuardFixed = true)); + } + } +} + +// The d7c0ec9 regression, pre-fix: the rehandshake teardown does NOT guard +// against consumed session states. The checker MUST find the assertion (a +// crash / unhandled rejection in TS) — check.sh runs this test expecting a +// bug to be found. +machine MainD7c0ec9Regression { + start state Init { + entry { + new Orchestrator((n = 3, dropBudget = 2, dropMix = 1, restartBudget = 0, + heartbeatBudget = 0, rehandshakeBudget = 2, corruptBudget = 0, + expectHonest = true, consumedGuardFixed = false, + zeroStateGuardFixed = true)); + } + } +} + +// The zero-state duplicate-delivery regression, pre-fix: the server does not +// honor the handshake's isReconnect flag, so a zero-state reconnect to a +// server that lost the session (restart) is accepted as new and the client's +// replay re-delivers. The checker MUST find the strengthened +// ExactlyOnceInOrder violation — check.sh runs this expecting a bug. +machine MainZeroStateDup { + start state Init { + entry { + new Orchestrator((n = 3, dropBudget = 1, dropMix = 0, restartBudget = 1, + heartbeatBudget = 0, rehandshakeBudget = 0, corruptBudget = 0, + expectHonest = true, consumedGuardFixed = true, + zeroStateGuardFixed = false)); + } + } +} + +module River = { ClientTransport, ServerTransport, WsConnection, EchoTimer, Orchestrator }; + +test tcHappyPath [main=MainHappyPath]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainHappyPath }); + +test tcDrops [main=MainDrops]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainDrops }); + +test tcRestart [main=MainRestart]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainRestart }); + +test tcByzantine [main=MainByzantine]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainByzantine }); + +test tcPhantom [main=MainPhantom]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainPhantom }); + +test tcRehandshake [main=MainRehandshake]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainRehandshake }); + +test tcD7c0ec9Regression [main=MainD7c0ec9Regression]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainD7c0ec9Regression }); + +test tcZeroStateDup [main=MainZeroStateDup]: + assert ExactlyOnceInOrder, AllCallsResolve, SessionIdPreserved, CleanShutdown in + (union River, { MainZeroStateDup }); diff --git a/verification/p/README.md b/verification/p/README.md new file mode 100644 index 00000000..e92ded59 --- /dev/null +++ b/verification/p/README.md @@ -0,0 +1,270 @@ +# P model of the River transport/session layer + +This directory holds a formal model of River's session layer — the seq/ack +exactly-once bookkeeping, send-buffer retransmission, the 4-case handshake, +transparent vs hard reconnects, session grace periods, the heartbeat watchdog, +and the rehandshake (credential refresh) exchange — written in the +[P language](https://p-org.github.io/P/) and explored with the P model +checker. The checker systematically explores interleavings of the client, the +server, and a fault-injecting WebSocket model, checking the invariants that +`PROTOCOL.md` promises and that `__tests__/properties/README.md` samples with +random fault schedules. + +## Running + +``` +nix develop .#verification # p (P CLI from NuGet), .NET SDK, uclid + z3 + # (PVerifier proofs), JDK 17 + maven (PEx) +verification/p/check.sh # compile + run every scenario + the inductive + # proof (~10 minutes) +``` + +or `npm run model:check`. This is intentionally not wired into `npm test`/CI. +Everything is packaged natively in Nix (see `flake.nix`: the `p` dotnet tool, +and UCLID5 built from a pinned master commit with the z3 4.12 Java bindings — +the released uclid 0.9.5 predates the `datatype` syntax PVerifier emits). + +`p check -tc -s ` runs one scenario; counterexample traces +land in `PCheckerOutput/BugFinding/`. `p check --list-tests` lists scenarios. +`p check --mode pex` (after `p compile --mode pex`) runs the exhaustive +JVM-based explorer instead of randomized bugfinding. + +## What is modeled + +| P machine | Mirrors | +| ------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `ClientTransport` (`PSrc/Client.p`) | the client session state machine (`transport/sessionStateMachine/`): Idle → Connecting → Handshaking → Connected (+ Dead, Done). Carries sessionId/seq/ack/seqSent/sendBuffer across reconnects; absolute-deadline grace period; retry budget; fresh-session-on-retriable-rejection. | +| `ServerTransport` (`PSrc/Server.p`) | `transport/server.ts`: the 4 handshake connect cases (transparent reconnect with both "in the future" rejections, hard reconnect, unknown session, new session), grace, restart (total state loss), active heartbeats, server-initiated rehandshake with async validate. | +| `WsConnection` (`PSrc/Network.p`) | one machine per WebSocket connection. P's per-machine-pair FIFO gives in-order reliable delivery within a connection for free; distinct connections race naturally (old connection's frames arriving around a new handshake). Faults: drop notifying both/one/neither side; the unnotified side is eventually unstuck by the watchdog (untimed C9b). | +| `Orchestrator` (`PTst/Testscript.p`) | the test driver: issues N calls (the server echoes each accepted payload — a stand-in for an rpc response), injects budgeted drops/restarts/heartbeats/rehandshakes at nondeterministic points, shuts down once every call resolved. | + +Abstractions: payloads are unique monotonic ints; codecs/serialization are out +of scope (covered by the A-series property tests); all durations are +nondeterministic orderings (timers fire via generation-guarded events with +budgeted defers, so both early and late firings are explored). The stream +layer is a thin overlay: payloads are assigned to streams in blocks of +`streamWidth`, the first message of a stream carries the StreamOpenBit, the +last carries the sender's half-close, and both endpoints keep per-stream +lifecycle records that live exactly as long as the session. + +Modeled bit-exactly: the three-way seq/ack receive comparison (`< ack` drop +duplicate, `> ack` invariant-violation/close-connection, `== ack` accept with +`ack = seq+1` and buffer filter `>= msg.ack`), byte-identical replay (a +replayed message carries its original, possibly stale ack), `seqSent` +send-ordering, `expectedSessionState` comparison, heartbeats consuming seqs, +and the consumed-session-state (linear handle) discipline around the +rehandshake's async gaps. + +## Properties + +| Check | Property catalog | How | +| ------------------------------------------------------------------------------------------------ | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| inline asserts (`assertBufferWindow`, send-ordering, `seq > ack` unreachable among honest peers) | C2, C3, C4 | the model self-checks on every send/accept/replay, mirroring the TS `invariant-violation` log oracle | +| `ExactlyOnceInOrder` (`PSpec/Delivery.p`) | C1 | between session resets — hence across any schedule of transparent reconnects — each side accepts payloads exactly once, in order | +| `AllCallsResolve` (`PSpec/Delivery.p`) | C5 + liveness | every issued call resolves exactly once (value or `UNEXPECTED_DISCONNECT`); hot state ⇒ an execution that quiesces with a hanging call is a bug | +| `SessionIdPreserved` (`PSpec/Reconnect.p`) | C6 | transparent reconnect keeps the session id | +| `CleanShutdown` (`PSpec/Reconnect.p`) | C7 | after shutdown starts, both endpoints eventually close | +| watchdog detection (deferred close to unnotified sides) | C9b (untimed) | phantom/one-sided drops must not hang the system; mutation M5 (watchdog disabled) makes `tcPhantom` fail with a liveness bug | +| stream lifecycle inline asserts (`trackRequestStream`, `trackResponseStream`) | B-series | open-exactly-once per stream instance, no data before open, no writes after the writer's half-close on either side; server may keep writing after the CLIENT's half-close (that is the point of half-close) | +| d7c0ec9 inline assert (`PSrc/Server.p`) | consumed-handle race | see below | +| `verified/SessionReconnect.p` inductive proof | C1/C2/C3 + reconnect, unbounded | see below | + +C9a (a live peer is never falsely killed by elapsed time) is a wall-clock +property outside an untimed model. + +## Scenarios + +| Test | Faults | Notes | +| --------------------- | ------------------------------------------------ | ------------------------------------------------------------------------------ | +| `tcHappyPath` | none | seq/ack + handshake + heartbeat sanity | +| `tcDrops` | ≤3 drops (both/client-only) | the core transparent-reconnect test | +| `tcRestart` | 1 server restart + ≤1 drop | hard reconnect, unknown-session rejection, `UNEXPECTED_DISCONNECT` | +| `tcByzantine` | misbehaving client injects duplicate/future seqs | duplicates silently dropped; future seq closes the connection, not the session | +| `tcPhantom` | ≤2 drops incl. server-only/silent | watchdog-driven recovery | +| `tcRehandshake` | ≤2 credential refreshes racing ≤2 drops | fixed (post-d7c0ec9) teardown guard: green | +| `tcD7c0ec9Regression` | same, PRE-fix teardown | **must find a bug**: `check.sh` asserts the checker reproduces the crash | + +## Findings + +1. **Duplicate delivery across server-side session loss in the zero-state + window — found by the model, confirmed against the implementation, and + FIXED.** If the server accepts messages but every ack/echo back to the + client is lost, and the server then loses the session (grace expiry or + restart) before the client reconnects, the client's handshake presents + `nextSentSeq = 0, nextExpectedSeq = 0` — indistinguishable from a new + session. The server accepted it (connect case 4), the client believed the + reconnect was transparent and replayed its buffer, and server handlers + executed the same requests a second time while the original calls hung + forever. `__tests__/zerostate.test.ts` reproduced this deterministically + against the TypeScript implementation. The fix: the client marks + reconnection attempts with `expectedSessionState.isReconnect` (an + optional, wire-compatible field — the session tracks `hadConnection` + across transitions), and the server rejects a marked reconnect to a + session it does not have with `SESSION_STATE_MISMATCH`, yielding the + documented hard-reconnect semantics (`UNEXPECTED_DISCONNECT`, fresh + session, no replay). With the fix, the `ExactlyOnceInOrder` monitor is + strengthened: the delivery watermark may only move FORWARD across session + resets (gaps allowed, re-delivery never), and all scenarios pass; the + pre-fix protocol lives on behind `zeroStateGuardFixed = false` as the + `tcZeroStateDup` regression, which `check.sh` requires the checker to + flag. + +2. **The d7c0ec9 consumed-handle race reproduces.** The model gives the + server's Connected state an instance epoch (the TS linear-typed state + proxy) and models the rehandshake's async `validate()` as a deferred + completion carrying the captured epoch. With the pre-fix teardown (reads + `session.to` before checking `_isConsumed`), the checker finds the crash: + a rehandshake rejection completing after a connection drop consumed the + state instance. With the fixed guard it is silent, and cleanup falls to + the replacement session — the model-level regression test for that bug + class. + +## The inductive proof (`verified/SessionReconnect.p`) + +The bounded scenarios above sample schedules; the proof covers unbounded +executions — and it covers the part of the protocol where the real bug +lived. `verified/SessionReconnect.p` models the delivery core (consecutive +seq assignment, retransmission from the unacked window, cumulative acks, the +`seq == ack` gate) PLUS sessions, total server state loss, hard client +resets, and the handshake with the `isReconnect` guard, over PVerifier's +network abstraction (arbitrary drop/duplication/reordering — strictly more +hostile than WebSocket connections, so the result holds a fortiori). + +`p compile -pf SessionReconnect.p -pn RiverSessionReconnect -md verification` +discharges 40 obligations with UCLID5 + Z3. **The theorem: across any +schedule of server restarts, session grace expiries, client hard resets, +replays, and reconnect handshakes, the application-level delivered stream +never regresses — no payload is ever delivered twice — and deliveries within +one session are gap-free.** The proof's load-bearing chain is the fix +itself: data only exists for sessions that were granted a handshake +(`msgs_granted`), a fresh (isReconnect=false) handshake is unique per +session and only in flight for never-granted sessions (`fresh_unique`, +`fresh_not_granted`), so accepting a session as NEW is safe, and every +deliverable in-flight message stays above the delivery watermark +(`deliverable_above_watermark`). Removing the `isReconnect` guard from the +model makes exactly that chain fail induction (mutation P2), and letting the +receiver accept `seq <= ack` breaks the window obligations (mutation P1) — +the proof is load-bearing on both the fix and the seq/ack gate. + +Model/implementation correspondence: the proved guards match the shipped code +line for line — the server's unknown-session rejection +(`clientNextSentSeq > 0 || isReconnect === true`, transport/server.ts), the +client-in-future check, fresh-session-on-retriable-rejection +(transport/client.ts), `hadConnection` set on entering Connected +(transitions.ts), cumulative-ack buffer pruning, and byte-identical +retransmission. Deliberate abstractions: data flows one direction (the +reverse direction holds by symmetry but is not separately machine-checked); +acks are standalone events rather than piggybacked (a superset of real +schedules); messages are composed at transmit time (the TS unsent-buffer is +invisible on the wire, so reachable wire states coincide); session tags on +wire events stand in for connection binding; and a session gets one fresh +handshake attempt (TS retries the same session, which is safe because +handshakes are connection-scoped — a requirement the proof surfaced and +PROTOCOL.md now states). + +Modeling notes: PVerifier's network has no connections, so the model folds +"a handshake request dies with its connection" into +one-fresh-attempt-per-session (a never-connected session retries by +resetting, which reaches the same states up to session renumbering); the +`granted` ghost set on the server is proof bookkeeping and persists across +modeled restarts. PVerifier subset notes: `choose()` and `$` inside compound +conditions are unsupported; proof commands must form a DAG (mutually +dependent invariants live in one `Lemma` group); the P CLI invokes +`uclid -M` (auto-inferred modifies sets). + +## Runtime conformance (PObserve) + +The model checker and the proof explore the _model_; PObserve closes the loop +by checking the _implementation_: `observe.sh` (or `npm run model:observe`) +runs the hegel property-based suite with execution tracing and replays the +traces through P spec machines — the same specification language as the +model, checked against reality. + +- **Trace tap** (`testUtil/fixtures/trace.ts`, activated by + `RIVER_TRACE_DIR`, zero library changes): a wrapper codec records every + outbound frame at encode time, `bindLogger` records every accepted inbound + frame (the `received msg` debug line carries seq/ack/streamId/sessionId) + plus out-of-order receives and `invariant-violation` lines, and the + transport events record session lifecycle. One JSONL file per generated + test case, ordered by a process-monotonic counter. +- **Specs** (`PObs/RiverTraceSpecs.p`, compiled with + `p compile --mode pobserve`): PObserve partitions the event stream per key + and runs one monitor instance per partition, so "per session" and "per + stream" scoping falls out of key choice (`observe.sh` runs one PObserve + pass per spec): + + | Spec | Partition | Mirrors | + | ------------------------- | ---------------------- | ------------------------------------------------------------------------------------------ | + | `AcceptedSeqContiguous` | session, receiver side | C1/C2: accepted seqs are exactly 0,1,2,... within a session, across transparent reconnects | + | `EncodedSeqDense` | run, sender side | assertSendOrdering's shadow: dense seq assignment | + | `SessionStateConformance` | side, session | real `sessionTransition` events follow the model's state graph exactly | + | `StreamFlagDiscipline` | session, stream, side | B-series: open-exactly-once, nothing after the sender's close | + | `NoInvariantViolations` | global | C4 + out-of-order receives never happen over reliable connections | + +- **Parser** (`PObs/parser/RiverTraceParser.kt`, Kotlin): JSONL → typed P + events; mode-selected via `--parserConfiguration`. Built by `observe.sh` + (kotlinc from the dev shell, stdlib bundled via `-include-runtime`) against + the Nix-packaged PObserve runtime (`packages..pobserve`, compiled + from the P repo's plain-Java sources with pinned Maven jars — the + artifacts are not on Maven Central). The port was verified equivalent to + the original Java parser: both run over the same 193-file trace corpus + with identical per-spec event/key/partition counts and identical + violation reports on poisoned traces. + +Current status: **all five specs pass** over the full property suite +(~7,000 records, ~380 sessions, ~480 streams per run). Detection was +validated with a poisoned trace (a seq gap makes `AcceptedSeqContiguous` +report the exact partition and assertion, with a replay window of the +preceding events) — and by a harness bug the pipeline caught during +bring-up: partition keys initially spanned test processes, and +`EncodedSeqDense` flagged the interleaving immediately. + +## Mutation validation + +Monitors were validated by seeding bugs and confirming detection +(re-run by hand; each mutation is a one-line change): + +| Mutation | Detected by | +| ------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------- | +| M1: buffer filter prunes `<= ack` instead of `< ack` (drops an unacked message) | `assertBufferWindow` | +| M2: skip send-buffer replay on reconnect | `AllCallsResolve` liveness (calls hang) | +| M3: duplicates reach the handler instead of being dropped | `ExactlyOnceInOrder` (needs `--sch-feedback`) | +| M4: transparent adopt forgets the inherited ack | `ExactlyOnceInOrder` (needs `--sch-feedback`) | +| M5: watchdog disabled | `AllCallsResolve` liveness in `tcPhantom` | +| M6: stale per-stream records kept across a new server session | NOT detected within 100k schedules — reaching it needs the rare zero-state hard-reconnect window; the stream asserts remain as defense-in-depth | +| P1 (proof): receiver accepts `seq <= ack` | 3 PVerifier obligations fail | + +Because M3/M4 were only found by the feedback scheduler, `check.sh` runs every +scenario under both the random and feedback strategies. + +## Model-design notes + +- **Timers**: generation-guarded fire events echoed by an `EchoTimer` machine, + with budgeted nondeterministic defers. This is the official P Timer idiom + (Tutorial/Common models a timer as a nondeterministic-delay self-loop) with + a bounded defer count instead of a probabilistic unbounded loop, and + generation guards instead of cancel messages (they encode the TS + absolute-deadline grace semantics directly). A naive immediate echo lands + the grace fire early in the owner's FIFO queue in most schedules, starving + the post-Connected part of the state space (this starvation was discovered + by reachability probes: transparent reconnect was unreachable before the + defer pattern). +- **Fault timing**: the injector decides to drop a connection when it is + created; the drop event's budgeted defers move the actual drop point across + the connection's message stream — the same shape as the hegel + write-schedule generator's `null` fault points. +- **Known model/TS divergences** (intentional): backoff durations, connect and + handshake timeouts are collapsed into the nondeterministic scheduler; server + restart does not kill in-flight handshake requests (equivalent to a fast + reconnect); retry-budget exhaustion is terminal (`Dead`) rather than + time-restored. + +## Roadmap + +- StreamCancelBit (abrupt full-close) and cancel races in the stream overlay. +- A parameterized test sweep (`test param (dropBudget in [...], ...)`) to + replace the per-scenario Main machines. +- PEx exhaustive runs of the smaller scenarios as a nightly job. +- Extending the proof toward multi-message transparent adoption detail + (inherited nonzero seq/ack windows across adoption are currently exercised + by the checker; the proof models adoption at the handshake level). diff --git a/verification/p/RiverP.pproj b/verification/p/RiverP.pproj new file mode 100644 index 00000000..aae93d4a --- /dev/null +++ b/verification/p/RiverP.pproj @@ -0,0 +1,9 @@ + + RiverP + + ./PSrc/ + ./PSpec/ + ./PTst/ + + ./PGenerated/ + diff --git a/verification/p/check.sh b/verification/p/check.sh new file mode 100755 index 00000000..426045f9 --- /dev/null +++ b/verification/p/check.sh @@ -0,0 +1,72 @@ +#!/usr/bin/env bash +# Compile the River P model and run every checker scenario. +# +# Requires the `p` CLI (P model checker) and the .NET SDK: +# nix develop .#verification +# +# Usage: ./check.sh [schedules] (default 10000) +set -euo pipefail +cd "$(dirname "$0")" + +SCHEDULES="${1:-10000}" + +p compile + +# Scenarios that must be bug-free. Each runs under two search strategies: +# the default random scheduler and the feedback-mutation scheduler (which +# reaches the rarer transparent-reconnect and rehandshake interleavings — +# mutation testing showed some seeded bugs are only found by feedback). +GREEN_TESTS=(tcHappyPath tcDrops tcRestart tcByzantine tcPhantom tcRehandshake) + +fail=0 +for t in "${GREEN_TESTS[@]}"; do + for strat in "" "--sch-feedback"; do + echo "--- $t $strat" + # shellcheck disable=SC2086 + # `p check` exits nonzero when it finds a bug; don't let set -e abort + out="$(p check -tc "$t" -s "$SCHEDULES" $strat 2>&1 | grep -E '\.\. Found [0-9]+ bug' | tail -1 || true)" + echo " $out" + if ! grep -q 'Found 0 bugs' <<<"$out"; then + echo " FAIL: $t found a bug (trace in PCheckerOutput/BugFinding/)" + fail=1 + fi + done +done + +# Regression scenarios that must FIND a bug. If the checker stops finding +# them, the model lost the races that motivated the fixes. +echo "--- tcZeroStateDup (expects a bug: pre-isReconnect duplicate delivery)" +out="$(p check -tc tcZeroStateDup -s $((SCHEDULES * 2)) --sch-feedback 2>&1 | grep -E '\.\. Found [0-9]+ bug' | tail -1 || true)" +echo " $out" +if grep -q 'Found 0 bugs' <<<"$out"; then + echo " FAIL: the zero-state duplicate-delivery regression no longer reproduces" + fail=1 +fi + +echo "--- tcD7c0ec9Regression (expects a bug: pre-fix consumed-handle crash)" +out="$(p check -tc tcD7c0ec9Regression -s $((SCHEDULES * 4)) --sch-feedback 2>&1 | grep -E '\.\. Found [0-9]+ bug' | tail -1 || true)" +echo " $out" +if grep -q 'Found 0 bugs' <<<"$out"; then + echo " FAIL: the d7c0ec9 regression no longer reproduces" + fail=1 +fi + +# Inductive proof (PVerifier -> UCLID5 -> Z3; provided by the .#verification +# dev shell). Unlike the bounded scenarios above, this covers UNBOUNDED +# executions: the seq/ack sliding window PLUS sessions, server state loss, +# handshakes, and reconnection — proving the isReconnect guard makes +# duplicate delivery unreachable (removing the guard from the model makes +# the proof fail). +echo "--- SessionReconnect inductive proof" +proof_out="$(cd verified && rm -rf PGenerated && p compile -pf SessionReconnect.p -pn RiverSessionReconnect -md verification 2>&1 || true)" +echo "$proof_out" | grep -E '🎉|❌' | sed 's/^/ /' +if echo "$proof_out" | grep -q '❌'; then + echo " FAIL: inductive proof did not go through" + fail=1 +fi + +if [ "$fail" -ne 0 ]; then + echo "MODEL CHECK FAILED" + exit 1 +fi +echo "MODEL CHECK PASSED" diff --git a/verification/p/observe.sh b/verification/p/observe.sh new file mode 100755 index 00000000..c95040cd --- /dev/null +++ b/verification/p/observe.sh @@ -0,0 +1,79 @@ +#!/usr/bin/env bash +# Runtime conformance: run the hegel property-based suite with execution +# tracing, then check the traces against the P model's invariants with +# PObserve (compiled from the same spec language as the model checker). +# +# Requires the verification dev shell: nix develop .#verification +# (provides `p`, `pobserve` + POBSERVE_HOME, and a JDK; the trace tap in +# testUtil/fixtures/trace.ts activates via RIVER_TRACE_DIR) +# +# Usage: ./observe.sh [trace-dir] +# With no argument, runs the property suite to produce fresh traces. +# With an argument, checks an existing trace directory. +set -euo pipefail +cd "$(dirname "$0")" + +REPO_ROOT="$(cd ../.. && pwd)" +TRACES="${1:-}" +if [ -n "$TRACES" ]; then + TRACES="$(cd "$OLDPWD" 2>/dev/null && cd "$(dirname "$TRACES")" && pwd)/$(basename "$TRACES")" || TRACES="$1" +fi + +# 1. compile the trace specs to Java monitors, then the Kotlin parser +p compile -pf PObs/RiverTraceSpecs.p -pn RiverTrace -md pobserve -o PObs/PGenerated +CP="$(cat "$POBSERVE_HOME/share/java/classpath.txt")" +rm -rf PObs/build && mkdir -p PObs/build +javac -cp "$CP" -d PObs/build PObs/PGenerated/PObserve/*.java +jar cf PObs/river-trace.jar -C PObs/build . +# -include-runtime bundles the Kotlin stdlib so the PObserve CLI's jar +# classloader can resolve it alongside the parser +kotlinc -cp "$CP:PObs/build" PObs/parser/RiverTraceParser.kt \ + -include-runtime -d PObs/river-parser.jar 2> >(grep -v '^warning:' >&2 || true) + +# 2. produce traces from the real implementation (unless given a directory) +if [ -z "$TRACES" ]; then + TRACES="$PWD/PObs/traces" + rm -rf "$TRACES" + echo "--- running the property suite with tracing (RIVER_TRACE_DIR)" + (cd "$REPO_ROOT" && RIVER_TRACE_DIR="$TRACES" npx vitest run \ + __tests__/properties/session.property.test.ts \ + __tests__/properties/streams.property.test.ts) +fi +echo "--- traces: $(ls "$TRACES" | wc -l) files, $(cat "$TRACES"/* | wc -l) records" + +# 3. check every spec (each partitions the trace stream differently) +declare -A MODES=( + [AcceptedSeqContiguous]=session + [EncodedSeqDense]=side + [SessionStateConformance]=machine + [StreamFlagDiscipline]=stream + [NoInvariantViolations]=global +) + +fail=0 +for spec in "${!MODES[@]}"; do + out="PObs/out/$spec" + rm -rf "$out" && mkdir -p "$out" + pobserve --jars PObs/river-trace.jar PObs/river-parser.jar --spec "$spec" \ + --parser RiverTraceParser --parserConfiguration "${MODES[$spec]}" \ + -l "$TRACES" -o "$out" > "$out/stdout.log" 2>&1 || true + summary="$(awk '/Total Events Read/{getline; getline; print; exit}' "$out/PObserveMetrics.txt" 2>/dev/null | tr -s ' ' || true)" + echo "--- $spec (${MODES[$spec]}): read/verified/keys/partitions: ${summary:-}" + violations=$(find "$out" -name 'replayEvents_*' 2>/dev/null | wc -l || true) + parser_errors=$(find "$out" -name 'ParserError*' -size +0 2>/dev/null | wc -l || true) + if [ "$violations" -gt 0 ]; then + echo " CONFORMANCE VIOLATION(S): $violations partition(s), see $out/replayEvents_*" + grep -m1 -h 'errorMessage=' "$out"/replayEvents_* | sed 's/^/ /' + fail=1 + fi + if [ "$parser_errors" -gt 0 ]; then + echo " PARSER ERRORS: see $out/" + fail=1 + fi +done + +if [ "$fail" -ne 0 ]; then + echo "RUNTIME CONFORMANCE FAILED: the implementation diverged from the model" + exit 1 +fi +echo "RUNTIME CONFORMANCE PASSED" diff --git a/verification/p/verified/SessionReconnect.p b/verification/p/verified/SessionReconnect.p new file mode 100644 index 00000000..46a4107f --- /dev/null +++ b/verification/p/verified/SessionReconnect.p @@ -0,0 +1,347 @@ +/***************************************************************************** +Inductive proof that the isReconnect handshake guard closes the zero-state +duplicate-delivery window — for ALL executions (PVerifier: UCLID5 + Z3). + +Beyond the seq/ack sliding window inside one session, this model covers where +the real bug lived: sessions, server state loss, handshakes, and +reconnection. + +The client owns a current session (monotonically numbered `sid`), issues +globally-unique payloads (`base + seqn + 1`, with `base` ratcheting past every +issued payload on a hard reset — abandoned messages are never re-issued), and +may only transmit while connected. Handshake requests carry the session id, +the send-buffer head (`nextSentSeq`), and `isReconnect` — whether this session +was ever connected (the fix shipped in transport/: the session's +`hadConnection` bit). The server holds at most one session, may lose ALL +state at any moment (restart / grace expiry), accepts data only for its +current session at exactly `seqn == ackNum`, and handles handshakes with the +FIXED rule: a request for an unknown session is rejected when +`nextSentSeq > 0 || isReconnect` — the zero-state window is exactly the case +where only `isReconnect` distinguishes a doomed replay from a fresh session. + +THE THEOREM (the in-handler asserts): the served application stream never +regresses — every delivered payload is strictly greater than the delivery +watermark W (no duplicate delivery, ever, across any number of resets, +losses, and reconnects), and consecutive deliveries within one session are +gap-free (payload == W + 1). + +The proof leans on one load-bearing chain, which is the fix itself: + data is only ever sent on a session that has connected (I6, lastConnSid) + => a handshake with isReconnect == false is only in flight for sessions + that never connected (I5) + => accepting a session as NEW is safe: no data for it exists anywhere + => every deliverable in-flight message is above the watermark (I10). +Deleting the isReconnect guard from handleHsReq makes I10 (and the theorem) +fail induction — the proof-level regression for this bug class. + +Run: p compile -pf SessionReconnect.p -pn RiverSessionReconnect -md verification +(inside `nix develop .#verification`) +*****************************************************************************/ + +event eMsg: (sid: int, seqn: int, payload: int); +event eAck: (sid: int, ack: int); +event eHsReq: (sid: int, nextSentSeq: int, isReconnect: bool); +event eHsOk: (sid: int); +event eHsReject: (sid: int); + +machine ClientCore { + var sid: int; // current session id, monotonic (init 1) + var base: int; // payload offset: this session's payloads are base+1.. + var seqNum: int; // next seq to assign in this session + var bufLo: int; // lowest unacked seq (send-buffer head) + var retry: int; // retransmit cursor (walks the window) + var connected: bool; // handshake completed on the current connection + var hadConn: bool; // this session ever connected (drives isReconnect) + var awaitingHs: bool; // at most one handshake request outstanding + var sentFreshReq: bool; // this session's one fresh (isReconnect=false) + // handshake has been sent — at most one ever exists + var lastConnSid: int; // ghost: highest session id that ever connected + + start state Run { + entry { + // one nondeterministic action per step ($ cannot appear in compound + // conditions in the verifier subset, hence the nested guards) + if ($) { + if (connected) { + // fresh send: globally-unique payload, bound to (sid, seqn) + send server(), eMsg, (sid = sid, seqn = seqNum, payload = base + seqNum + 1); + seqNum = seqNum + 1; + } + } else if ($) { + if (connected && bufLo < seqNum) { + // retransmit any unacked message byte-identically (buffer replay) + if (retry < bufLo || retry >= seqNum) { + retry = bufLo; + } + send server(), eMsg, (sid = sid, seqn = retry, payload = base + retry + 1); + if ($) { + retry = retry + 1; + } + } + } else if ($) { + if (!connected && !awaitingHs) { + if (hadConn || !sentFreshReq) { + // (re)connect. nextSentSeq is the send-buffer head, isReconnect + // is the hadConnection bit — the fix under proof. A session gets + // exactly ONE fresh (isReconnect=false) handshake: in reality a + // handshake request dies with its connection (sockets close, + // pending sessions time out), so a fresh request can never + // outlive the session's first connect; the untimed model folds + // that into one-fresh-attempt-per-session (a never-connected + // session that gives up resets instead, which reaches the same + // states up to session renumbering). + awaitingHs = true; + sentFreshReq = true; + send server(), eHsReq, (sid = sid, nextSentSeq = bufLo, isReconnect = hadConn); + } + } + } else if ($) { + if (connected) { + // connection drop observed by the client. The outstanding-attempt + // flag is NOT cleared here: a handshake request never outlives its + // connection attempt in reality (sockets die with the connection + // and pending sessions have a handshake timeout), which the model + // folds into at-most-one-outstanding-attempt per session. + connected = false; + } + } else if ($) { + // hard reset: grace expiry / fatal rejection. Every issued payload of + // this session is abandoned (resolved UNEXPECTED_DISCONNECT) and + // never re-issued: base ratchets past all of them. + newSession(); + } + if ($) { + goto Run; + } + } + on eAck do (a: (sid: int, ack: int)) { + // cumulative ack for the CURRENT session only (stale sessions' acks + // arrive after a reset and must not touch the new session's window) + if (a.sid == sid && a.ack > bufLo) { + bufLo = a.ack; + } + } + on eHsOk do (h: (sid: int)) { + if (h.sid == sid) { + connected = true; + hadConn = true; + awaitingHs = false; + if (lastConnSid < sid) { + lastConnSid = sid; + } + } + } + on eHsReject do (h: (sid: int)) { + if (h.sid == sid) { + // SESSION_STATE_MISMATCH: hard reset to a fresh session + newSession(); + } + } + } + + fun newSession() { + base = base + seqNum; + sid = sid + 1; + seqNum = 0; + bufLo = 0; + retry = 0; + connected = false; + hadConn = false; + awaitingHs = false; + sentFreshReq = false; + } +} + +machine ServerCore { + var hasSess: bool; + var cur: int; // adopted session id + var ackNum: int; // next expected seq for cur + var w: int; // ghost: watermark = highest payload delivered + var lastDelivSid: int; // ghost: session of the most recent delivery + var granted: set[int]; // ghost: sids ever sent an eHsOk + + start state Run { + entry { + if ($) { + if (hasSess) { + // total server state loss: restart or session grace expiry + hasSess = false; + } + } + if ($) { + goto Run; + } + } + on eHsReq do (h: (sid: int, nextSentSeq: int, isReconnect: bool)) { + if (hasSess && cur == h.sid) { + // transparent reconnect to the session we hold + if (h.nextSentSeq > ackNum) { + // client is in the future + send client(), eHsReject, (sid = h.sid,); + } else { + granted += (h.sid); + send client(), eHsOk, (sid = h.sid,); + } + } else if (h.nextSentSeq > 0 || h.isReconnect) { + // THE FIX: an unknown session is rejected not only when the seq + // counters are nonzero but also when the client marks the attempt as + // a reconnection — the zero-state window is exactly + // nextSentSeq == 0 && isReconnect == true + send client(), eHsReject, (sid = h.sid,); + } else { + // new session (hard reconnect implicitly deletes any old one) + hasSess = true; + cur = h.sid; + ackNum = 0; + granted += (h.sid); + send client(), eHsOk, (sid = h.sid,); + } + } + on eMsg do (m: (sid: int, seqn: int, payload: int)) { + if (hasSess && m.sid == cur && m.seqn == ackNum) { + // THE THEOREM: the application stream never regresses... + assert m.payload > w, + "duplicate delivery: a payload at or below the watermark reached the application"; + // ...and within one session it is gap-free + if (lastDelivSid == cur) { + assert m.payload == w + 1, + "delivery gap within a session"; + } + w = m.payload; + lastDelivSid = cur; + ackNum = ackNum + 1; + if ($) { + send client(), eAck, (sid = cur, ack = ackNum); + } + } + // otherwise: unknown/stale session or out-of-window seq — dropped + } + } +} + +/**************************** system configuration ***************************/ + +pure client(): machine; +pure server(): machine; + +init-condition forall (m: machine) :: m == client() <==> m is ClientCore; +init-condition forall (m: machine) :: m == server() <==> m is ServerCore; +init-condition forall (c: ClientCore) :: + c.sid == 1 && c.base == 0 && c.seqNum == 0 && c.bufLo == 0 && c.retry == 0 && + !c.connected && !c.hadConn && !c.awaitingHs && !c.sentFreshReq && + c.lastConnSid == 0; +init-condition forall (s: ServerCore) :: + !s.hasSess && s.cur == 0 && s.ackNum == 0 && s.w == 0 && s.lastDelivSid == 0 && + s.granted == default(set[int]); + +/******************************** invariants *********************************/ + +Lemma reconnect { + // configuration and routing + invariant one_client: forall (m: machine) :: m == client() <==> m is ClientCore; + invariant one_server: forall (m: machine) :: m == server() <==> m is ServerCore; + invariant no_msg_to_client: forall (e: eMsg, c: ClientCore) :: e targets c ==> !inflight e; + invariant no_hsreq_to_client: forall (e: eHsReq, c: ClientCore) :: e targets c ==> !inflight e; + invariant no_ack_to_server: forall (e: eAck, s: ServerCore) :: e targets s ==> !inflight e; + invariant no_hsok_to_server: forall (e: eHsOk, s: ServerCore) :: e targets s ==> !inflight e; + invariant no_hsreject_to_server: forall (e: eHsReject, s: ServerCore) :: e targets s ==> !inflight e; + + // client-local structure + invariant window_wf: forall (c: ClientCore) :: 0 <= c.bufLo && c.bufLo <= c.seqNum; + invariant base_nonneg: forall (c: ClientCore) :: c.base >= 0; + invariant sid_positive: forall (c: ClientCore) :: c.sid >= 1; + invariant conn_ghost: forall (c: ClientCore) :: + (c.hadConn ==> c.lastConnSid == c.sid) && + (!c.hadConn ==> c.lastConnSid < c.sid); + invariant unconnected_fresh: forall (c: ClientCore) :: + !c.hadConn ==> c.seqNum == 0 && c.bufLo == 0; + invariant connected_implies_had: forall (c: ClientCore) :: c.connected ==> c.hadConn; + + // wire: current-session data is bound to the window and the payload line + invariant cur_msg_binding: forall (e: eMsg, c: ClientCore) :: + inflight e && e.sid == c.sid ==> + e.payload == c.base + e.seqn + 1 && 0 <= e.seqn && e.seqn < c.seqNum; + // wire: stale-session data is from the past — below the base ratchet + invariant old_msg_bounded: forall (e: eMsg, c: ClientCore) :: + inflight e && e.sid != c.sid ==> e.sid < c.sid && e.payload <= c.base; + // data only ever exists for sessions that connected (transmit gating) + invariant sent_needs_connection: forall (e: eMsg, c: ClientCore) :: + inflight e ==> e.sid <= c.lastConnSid; + // a handshake claiming "not a reconnect" carries a zero send-buffer head + // (an unconnected session has never transmitted) + invariant fresh_hsreq_honest: forall (e: eHsReq, c: ClientCore) :: + inflight e && !e.isReconnect ==> e.nextSentSeq == 0; + // ...and there is at most one such request per session, ever (one-shot) + invariant fresh_unique: forall (e1: eHsReq, e2: eHsReq) :: + inflight e1 && inflight e2 && !e1.isReconnect && !e2.isReconnect && + e1.sid == e2.sid ==> e1 == e2; + invariant fresh_oneshot: forall (e: eHsReq, c: ClientCore) :: + inflight e && !e.isReconnect && e.sid == c.sid ==> c.sentFreshReq && !c.hadConn; + invariant true_req_had: forall (e: eHsReq, c: ClientCore) :: + inflight e && e.isReconnect && e.sid == c.sid ==> c.hadConn; + // grant bookkeeping: an eHsOk exists only for granted sids, a granted sid + // has consumed its fresh request, and grants never outrun the client + invariant hsok_granted: forall (r: eHsOk, s: ServerCore) :: + inflight r ==> r.sid in s.granted; + invariant fresh_not_granted: forall (e: eHsReq, s: ServerCore) :: + inflight e && !e.isReconnect ==> !(e.sid in s.granted); + invariant granted_known: forall (c: ClientCore, s: ServerCore) :: + c.sid in s.granted ==> c.sentFreshReq || c.hadConn; + invariant granted_bounded: forall (c: ClientCore, s: ServerCore, x: int) :: + x in s.granted ==> x <= c.sid; + // the granted ghost is the connection witness: data, connected state, the + // held session, reconnect-flagged requests, and past deliveries all imply + // membership; a fresh request implies NON-membership (fresh_not_granted), + // so accepting a session as new is provably safe + invariant msgs_granted: forall (e: eMsg, s: ServerCore) :: + inflight e ==> e.sid in s.granted; + invariant connected_granted: forall (c: ClientCore, s: ServerCore) :: + c.hadConn ==> c.sid in s.granted; + invariant cur_granted: forall (s: ServerCore) :: + s.hasSess ==> s.cur in s.granted; + invariant true_req_granted: forall (e: eHsReq, s: ServerCore) :: + inflight e && e.isReconnect ==> e.sid in s.granted; + invariant deliv_granted: forall (s: ServerCore) :: + s.lastDelivSid == 0 || s.lastDelivSid in s.granted; + invariant hsreq_sid_bounded: forall (e: eHsReq, c: ClientCore) :: + inflight e ==> e.sid <= c.sid; + + // acks describe real receiver progress of the session they name + invariant ack_bounded: forall (e: eAck, c: ClientCore, s: ServerCore) :: + inflight e && e.sid == c.sid ==> e.ack <= c.seqNum; + + // server/client window relations while the server holds the live session + invariant recv_le_sent: forall (c: ClientCore, s: ServerCore) :: + s.hasSess && s.cur == c.sid ==> 0 <= s.ackNum && s.ackNum <= c.seqNum; + invariant watermark_le_issued: forall (c: ClientCore, s: ServerCore) :: + s.w <= c.base + c.seqNum; + invariant watermark_cur: forall (c: ClientCore, s: ServerCore) :: + s.hasSess && s.cur == c.sid ==> s.w <= c.base + s.ackNum; + invariant deliv_ghost: forall (c: ClientCore, s: ServerCore) :: + s.lastDelivSid <= c.lastConnSid; + invariant watermark_cur_tight: forall (c: ClientCore, s: ServerCore) :: + s.hasSess && s.cur == c.sid && s.lastDelivSid == s.cur ==> + s.w == c.base + s.ackNum; + + // messages of one session lie on one payload line (retransmits are + // byte-identical), so deliveries on a stale session stay ordered too + invariant same_sid_linear: forall (e1: eMsg, e2: eMsg) :: + inflight e1 && inflight e2 && e1.sid == e2.sid ==> + e1.payload - e1.seqn == e2.payload - e2.seqn; + // if the last delivery was on the currently-held session, the watermark + // sits exactly at that session's line at ackNum (stale or current) + invariant line_tight: forall (e: eMsg, s: ServerCore) :: + inflight e && s.hasSess && e.sid == s.cur && s.lastDelivSid == s.cur ==> + s.w == e.payload - e.seqn + s.ackNum - 1; + + // THE MASTER INVARIANT: anything the server could deliver next is above + // the watermark — duplicates are unreachable + invariant deliverable_above_watermark: forall (e: eMsg, s: ServerCore, c: ClientCore) :: + inflight e && s.hasSess && e.sid == s.cur && e.seqn >= s.ackNum ==> + e.payload > s.w; +} + +Proof { + prove reconnect; + prove default using reconnect; +}