fix: stabilize hosted destroyer scenarios

.github/workflows/destroyer.yml

@@ -112,6 +112,11 @@ jobs:
           - schedule-due-storm-isolation
           - same-shard-family-fairness
           - actor-supervision-failpoint
+          - family-actor-partial-failure-isolation
+          - same-shard-family-failure-isolation
+          - family-actor-exhaustion-readiness
+          - family-actor-degradation-observability
+          - family-actor-inflight-concurrent-failure
     steps:
       - name: Check out Fitz Destroyer
         uses: actions/checkout@v7

README.md

@@ -163,9 +163,9 @@ scales raise the due set to at least 2,000 and 5,000 definitions respectively.
 family-actor shard, continuously fills one family's Notice lane, and requires
 every sibling-family delivery canary to complete within the request timeout.
 
-`same-shard-family-failure-isolation` pins authenticated families 1 and 9 to
-the same one of eight family-actor shards, then panics family 1's Stream and RPC
-actors. The failed family must reject, family 9 must keep progressing without
+`same-shard-family-failure-isolation` pins authenticated families 1 and 5 to
+the same one of four family-actor shards, then panics family 1's Stream and RPC
+actors. The failed family must reject, family 5 must keep progressing without
 cross-family delivery, and broker readiness must remain healthy.
 
 `family-actor-exhaustion-readiness` fails the two provisioned Stream families
@@ -321,14 +321,20 @@ fails the run; recovered session-cleanup retries are recorded in the artifacts.
 `domain-pressure` runs a short, continuously bombarding client fleet without
 injecting faults. Use `--domains` to isolate one domain or an interference pair.
 It requires every selected domain to make progress on every client in each
-ten-second window and fails on definite operation errors. Queue operations with
-an unknown durable outcome are accepted only when exact reconciliation proves
-that every deterministic sequence resolved at most once.
+ten-second window and fails on definite operation errors. A typed Queue 4005
+rejection is retried with bounded exponential backoff because the request was
+not accepted; each retry remains explicit in stage evidence. Queue operations
+with an unknown durable outcome are accepted only when exact reconciliation
+proves that every deterministic sequence resolved at most once.
 KV and Schedule repeatedly update one logical resource per client, while Stream
-appends to one resource per client. This keeps route and current-state cardinality
-bounded so RSS diagnostics are not dominated by intentionally abandoned KV keys
-or Schedule definitions; Stream history still grows according to its durable
-append-only contract.
+appends to one resource per client. Durable loops run at a steady maximum of one
+cycle per second per client; dedicated overload scenarios own saturation tests.
+This keeps route and current-state cardinality bounded so RSS diagnostics are not
+dominated by intentionally abandoned KV keys or Schedule definitions; Stream
+history still grows according to its durable append-only contract. Live-domain
+loops use a shorter cadence, including pacing Notice publication so its
+fire-and-forget loop cannot monopolize the shared family lane and starve the
+response-bearing domain probes.
 `pressure-evidence.json` contains per-client/domain/stage totals, latency
 percentiles, normalized error samples, Queue reconciliation, broker snapshots,
 and diagnostic warnings.

compose.destroyer.yml

@@ -27,6 +27,7 @@ services:
       - "127.0.0.1::9100"
 
   fitz:
+    cpus: "${FITZ_CPU_LIMIT:-0}"
     image: "${FITZ_IMAGE:-ghcr.io/cntryl/fitz:latest}"
     restart: "no"
     stop_grace_period: 15s
@@ -50,6 +51,7 @@ services:
       FITZ_METRICS_BIND_ADDR: 0.0.0.0
       FITZ_METRICS_PORT: "9090"
       FITZ_DRAIN_GRACE_SECONDS: "1"
+      FITZ_DESTROYER_FAILPOINTS: "${FITZ_DESTROYER_FAILPOINTS:-disabled}"
       FITZ_STORAGE_MODE: cloud
       FITZ_STORAGE_PROVIDER: sqrzl-s3
       FITZ_STORAGE_ENDPOINT: http://storage-proxy:9000

package.json

@@ -16,7 +16,7 @@
     "check": "npm run typecheck && npm test"
   },
   "dependencies": {
-    "@cntryl/fitz": "0.0.21"
+    "@cntryl/fitz": "0.0.22"
   },
   "devDependencies": {
     "@types/node": "^24.10.0",

src/family-shard-topology.ts

@@ -0,0 +1,8 @@
+export const DESTROYER_FAMILY_ACTOR_SHARD_COUNT = 4;
+export const DESTROYER_PRIMARY_FAMILY = 1;
+export const DESTROYER_SAME_SHARD_FAMILY = DESTROYER_FAMILY_ACTOR_SHARD_COUNT + 1;
+
+export const DESTROYER_FAMILY_ACTOR_FAMILIES = Array.from(
+  { length: DESTROYER_SAME_SHARD_FAMILY },
+  (_, index) => index + 1,
+);

src/orchestration/actor-supervision-failpoint.ts

@@ -24,6 +24,13 @@ export function assertActorSupervisionEvidence(record: Readonly<Record<string, u
   }
 }
 
+export function activeFaultObservationTimeoutMs(
+  requestTimeoutMs: number,
+  startupTimeoutMs: number,
+): number {
+  return Math.min(startupTimeoutMs, requestTimeoutMs * 3);
+}
+
 export async function runActorSupervisionFailpointScenario(stack: ComposeStack, config: RunConfig, shape: WorkloadShape, artifacts: Artifacts): Promise<void> {
   const startedAt = performance.now();
   let domainsInjected = 0;
@@ -97,7 +104,11 @@ export async function runActorSupervisionFailpointScenario(stack: ComposeStack,
   await stack.waitForAllClientDomains(activeFaultStartedAt, config.clientReplicas);
   const faultStartedAt = new Date();
   await injectAndWaitForReadinessWithdrawal("all-domain", config);
-  const activeFaultErrors = await stack.waitForAllClientErrors(faultStartedAt, config.clientReplicas);
+  const activeFaultErrors = await stack.waitForAllClientErrors(
+    faultStartedAt,
+    config.clientReplicas,
+    activeFaultObservationTimeoutMs(config.requestTimeoutMs, config.startupTimeoutMs),
+  );
   await stack.stopBombardClientsAndCapture("actor-supervision-active-fault");
   await stack.stopFitz();
   await stack.restartFitz();

src/orchestration/compose-jobs.ts

@@ -12,15 +12,13 @@ type ComposeRunner = (
 
 interface StorageFaultRecoveryOperations {
   stopFitz: () => Promise<void>;
-  restartStorage: () => Promise<void>;
   startFitz: () => Promise<void>;
 }
 
 export async function executeStorageFaultRecovery(
   operations: StorageFaultRecoveryOperations,
 ): Promise<void> {
   await operations.stopFitz();
-  await operations.restartStorage();
   await operations.startFitz();
 }
 

src/orchestration/compose.ts

@@ -1,6 +1,12 @@
 import { access } from "node:fs/promises";
 import { join } from "node:path";
 import type { RunConfig } from "../config.js";
+import {
+  DESTROYER_FAMILY_ACTOR_FAMILIES,
+  DESTROYER_FAMILY_ACTOR_SHARD_COUNT,
+  DESTROYER_PRIMARY_FAMILY,
+  DESTROYER_SAME_SHARD_FAMILY,
+} from "../family-shard-topology.js";
 import { type Domain, type WorkloadShape } from "../workloads/model.js";
 import { Artifacts } from "./artifacts.js";
 import { runCommand, type CommandResult } from "./command.js";
@@ -78,6 +84,8 @@ export class ComposeStack {
   #jobSequence = 0;
   #roleSequence = 0;
   #metricsUrl: string | undefined;
+  #pressureRssBytes: number | undefined;
+  #pressureRssSampledAt = 0;
 
   constructor(
     config: RunConfig,
@@ -121,10 +129,7 @@ export class ComposeStack {
             FITZ_ASSUME_EXTERNAL_TLS: "true",
             FITZ_JWT_HMAC_SECRET: "fitz-destroyer-local-auth-only",
             FITZ_JWT_AUDIENCES: "fitz-destroyer",
-            FITZ_ROUTE_FAMILIES: config.scenario === "same-shard-family-fairness" || config.scenario === "same-shard-family-failure-isolation" || config.scenario === "family-actor-inflight-concurrent-failure" ? "1,2,3,4,5,6,7,8,9" : "1,2",
-            FITZ_ROUTE_FAMILY_MAP: config.scenario === "family-actor-inflight-concurrent-failure" ? "identity-a=1,identity-b=2,identity-c=9" : config.scenario === "same-shard-family-fairness" || config.scenario === "same-shard-family-failure-isolation" ? "identity-a=1,identity-b=9" : "identity-a=1,identity-b=2",
-            FITZ_ROUTE_FAMILY_CLAIM: "tid",
-            ...(config.scenario === "same-shard-family-fairness" || config.scenario === "same-shard-family-failure-isolation" || config.scenario === "family-actor-inflight-concurrent-failure" ? { FITZ_CPU_LIMIT: "8" } : {}),
+            ...authenticatedRouteFamilyEnvironment(config.scenario),
           }
         : {}),
     };
@@ -366,20 +371,12 @@ export class ComposeStack {
     if (containers.length !== 1 || container === undefined) {
       throw new Error(`Expected one running Fitz container for pressure snapshot, found ${containers.length}`);
     }
-    const [queue, rpc, prometheus, stats] = await Promise.all([
+    const [queue, rpc, prometheus, rssBytes] = await Promise.all([
       this.fetchJson("/api/v1/all/queue/stats"),
       this.fetchJson("/api/v1/all/rpc/stats"),
       this.fetchTextAt(`${metricsUrl}/metrics`, "Prometheus /metrics"),
-      runCommand(
-        "docker",
-        ["stats", "--no-stream", "--format", "{{json .}}", container],
-        { cwd: this.#config.rootDir },
-      ),
+      this.pressureRss(container),
     ]);
-    const record = JSON.parse(stats.stdout.trim()) as { MemUsage?: unknown };
-    if (typeof record.MemUsage !== "string") {
-      throw new Error(`Docker stats omitted Fitz MemUsage: ${stats.stdout.trim()}`);
-    }
     return {
       timestamp: new Date().toISOString(),
       queue,
@@ -418,10 +415,29 @@ export class ComposeStack {
           "fitz_router_high_lane_backpressure_total",
         ),
       },
-      rssBytes: parseDockerMemoryUsage(record.MemUsage),
+      rssBytes,
     };
   }
 
+  private async pressureRss(container: string): Promise<number> {
+    const now = Date.now();
+    if (this.#pressureRssBytes !== undefined && now - this.#pressureRssSampledAt < 10_000) {
+      return this.#pressureRssBytes;
+    }
+    const stats = await runCommand(
+      "docker",
+      ["stats", "--no-stream", "--format", "{{json .}}", container],
+      { cwd: this.#config.rootDir },
+    );
+    const record = JSON.parse(stats.stdout.trim()) as { MemUsage?: unknown };
+    if (typeof record.MemUsage !== "string") {
+      throw new Error(`Docker stats omitted Fitz MemUsage: ${stats.stdout.trim()}`);
+    }
+    this.#pressureRssBytes = parseDockerMemoryUsage(record.MemUsage);
+    this.#pressureRssSampledAt = now;
+    return this.#pressureRssBytes;
+  }
+
   async prometheusMetricValue(name: string): Promise<number> {
     const metricsUrl = await this.metricsUrl();
     const prometheus = await this.fetchTextAt(`${metricsUrl}/metrics`, "Prometheus /metrics");
@@ -488,10 +504,6 @@ export class ComposeStack {
         "fitz-after-storage-exhaustion",
         true,
       ),
-      restartStorage: async () => {
-        await this.killAndRemoveService("sqrzl", "sqrzl-after-exhaustion", true);
-        await this.compose(["up", "-d", "--no-deps", "sqrzl"], { stream: true });
-      },
       startFitz: async () => {
         await this.compose(["up", "-d", "--no-deps", "--no-build", "fitz"], { stream: true });
         await this.waitReady();
@@ -646,8 +658,12 @@ export class ComposeStack {
     throw new Error(`Timed out waiting for fresh client success in every domain: ${lastStatus}`);
   }
 
-  async waitForAllClientErrors(since: Date, replicas: number): Promise<number> {
-    const deadline = Date.now() + this.#config.requestTimeoutMs;
+  async waitForAllClientErrors(
+    since: Date,
+    replicas: number,
+    timeoutMs = this.#config.requestTimeoutMs,
+  ): Promise<number> {
+    const deadline = Date.now() + timeoutMs;
     while (Date.now() < deadline) {
       const containers = await this.serviceContainers("client", true);
       if (containers.length === replicas) {
@@ -1069,3 +1085,29 @@ export class ComposeStack {
     );
   }
 }
+
+function authenticatedRouteFamilyEnvironment(
+  scenario: RunConfig["scenario"],
+): Readonly<Record<string, string>> {
+  if (scenario === "family-actor-inflight-concurrent-failure") {
+    return {
+      FITZ_ROUTE_FAMILIES: DESTROYER_FAMILY_ACTOR_FAMILIES.join(","),
+      FITZ_ROUTE_FAMILY_MAP: `identity-a=${DESTROYER_PRIMARY_FAMILY},identity-b=2,identity-c=${DESTROYER_SAME_SHARD_FAMILY}`,
+      FITZ_ROUTE_FAMILY_CLAIM: "tid",
+      FITZ_CPU_LIMIT: String(DESTROYER_FAMILY_ACTOR_SHARD_COUNT),
+    };
+  }
+  if (scenario === "same-shard-family-fairness" || scenario === "same-shard-family-failure-isolation") {
+    return {
+      FITZ_ROUTE_FAMILIES: DESTROYER_FAMILY_ACTOR_FAMILIES.join(","),
+      FITZ_ROUTE_FAMILY_MAP: `identity-a=${DESTROYER_PRIMARY_FAMILY},identity-b=${DESTROYER_SAME_SHARD_FAMILY}`,
+      FITZ_ROUTE_FAMILY_CLAIM: "tid",
+      FITZ_CPU_LIMIT: String(DESTROYER_FAMILY_ACTOR_SHARD_COUNT),
+    };
+  }
+  return {
+    FITZ_ROUTE_FAMILIES: "1,2",
+    FITZ_ROUTE_FAMILY_MAP: "identity-a=1,identity-b=2",
+    FITZ_ROUTE_FAMILY_CLAIM: "tid",
+  };
+}

src/orchestration/pressure.ts

@@ -125,11 +125,16 @@ export async function runPressureScenario(
   }
 
   try {
+    const verificationCompletedAt = pressureVerificationCompletedAtMs(
+      pressureStartedAt.getTime(),
+      pressureCompletedAt.getTime(),
+      requestedDurationMs,
+    );
     assertProgressWindows(
       clientLogs,
       config.bombardDomains,
       pressureStartedAt.getTime(),
-      pressureCompletedAt.getTime(),
+      verificationCompletedAt,
     );
   } catch (error) {
     assertionFailures.push(errorMessage(error));
@@ -284,6 +289,14 @@ export function assertProgressWindows(
   }
 }
 
+export function pressureVerificationCompletedAtMs(
+  startedAtMs: number,
+  completedAtMs: number,
+  requestedDurationMs: number,
+): number {
+  return Math.min(completedAtMs, startedAtMs + requestedDurationMs);
+}
+
 async function samplePressure(
   stack: ComposeStack,
   artifacts: Artifacts,
@@ -313,10 +326,10 @@ export function pressureUnexpectedErrors(
   return clients.flatMap((client) =>
     domains.flatMap((domain) => {
       const evidence = client.domains[domain];
-      // Queue enqueue/complete timeouts have an explicitly unknown durable
-      // outcome. They are correctness failures only when exact reconciliation
-      // fails; definite stage failures still fail the pressure run here.
-      const count = domain === "queue"
+      // Queue outcomes are reconciled after the run. Stream outcomes are
+      // reconciled inline against the latest committed offset. Definite stage
+      // failures still fail the pressure run here.
+      const count = domain === "queue" || domain === "stream"
         ? Object.values(evidence?.stages ?? {}).reduce(
             (total, stage) => total + stage.failed,
             0,
@@ -432,6 +445,9 @@ function parseStage(value: unknown, label: string): EvidenceStage {
     succeeded: numericValue(record.succeeded, `${label}.succeeded`),
     failed: numericValue(record.failed, `${label}.failed`),
     ambiguous: numericValue(record.ambiguous, `${label}.ambiguous`),
+    retryableBackpressure: record.retryableBackpressure === undefined
+      ? 0
+      : numericValue(record.retryableBackpressure, `${label}.retryableBackpressure`),
     expectedShutdownCancellations: {
       failed: numericValue(shutdownCancellations.failed, `${label} shutdown cancellation failures`),
       ambiguous: numericValue(shutdownCancellations.ambiguous, `${label} ambiguous shutdown cancellations`),
@@ -486,6 +502,7 @@ function emptyEvidenceStage(): EvidenceStage {
     succeeded: 0,
     failed: 0,
     ambiguous: 0,
+    retryableBackpressure: 0,
     expectedShutdownCancellations: { failed: 0, ambiguous: 0 },
     latencyHistogram,
     latency: latencySummary(latencyHistogram),
@@ -499,6 +516,7 @@ function mergeStage(target: EvidenceStage, source: EvidenceStage): void {
   target.succeeded += source.succeeded;
   target.failed += source.failed;
   target.ambiguous += source.ambiguous;
+  target.retryableBackpressure += source.retryableBackpressure;
   target.expectedShutdownCancellations.failed += source.expectedShutdownCancellations.failed;
   target.expectedShutdownCancellations.ambiguous += source.expectedShutdownCancellations.ambiguous;
   target.latencyHistogram = mergeLatencyHistograms([