Preserve measurement targets when liquidity publication is rejected

docs/dex-liquidity.md

@@ -55,7 +55,7 @@ Publication retention treats an unreferenced `staged` generation older than thre
 
 Persistence is heap-bounded without weakening that publication fence. Candidate rows stream through 15-row buffers packed into no more than five multi-row statements per D1 transaction; history and other scoring-side writes construct and execute at most 25 D1 statements at a time; later batches are not prepared until the prior batch resolves. The schema-v1 source handoff serializes and clears measured-target maps before the larger metric/pool graph, releases every consumed source map, uses direct-buffer UTF-8 encoding without a second line array or joined payload copy, and carries the already-loaded trust-filtered primary-price map into the scoring consumer. Price observations are derived one active asset at a time, staged in generation-keyed `dex_price_run_rows`, and exact-count validated before one atomic D1 batch checks an in-write current-generation fence and replaces the complete `dex_prices` table. The consumer then clears its primary-price and exact-observation graphs before challenger publication. Challenger payloads pack multiple projected pool rows into each statement up to D1's bind ceiling while consuming the retained-pool map entry by entry; after every payload row lands, one direct `json_each`-driven UPSERT atomically advances all complete asset snapshot pointers and derives `has_rows` from the durable payload. Superseded payload cleanup remains strictly after that pointer fence. A payload or pointer interruption therefore leaves the previous challenger snapshot set authoritative instead of exposing a mixed partial refresh. The stage remains intact across ambiguous D1 retries, is removed only after the public replacement is verified, and otherwise turns over after three hours through cleanup bounded to eight generations per run; cleanup protects the in-flight generation, active staged publication work, and the generation named by the public `__global__` row. Depth-stability values are likewise written only to the current generation's private `dex_liquidity_run_rows` rows before one generation-guarded atomic update reaches the public table. A staging or final-batch error propagates to the cron and leaves each public price/depth surface wholly on its previous generation. Retention cleanup is best-effort after publication: failures degrade telemetry but do not invalidate an otherwise successful publish, and every pass reports its cutoff, deleted count, oldest remaining row, duration, and error. Consumed scoring-stage rows are deleted by the next successful stage cleanup; abandoned writing, ready, or failed stages become eligible after two hours, while the current generation stays protected. Consumed pool/score maps are cleared as their downstream stages complete, and progress advances through generation, price, challenger, history, and depth substages so a platform interruption is attributable. After measured targets have been adjusted onto retained pools and captured for target publication, the producer target maps are cleared before any proof-heavy join evidence is loaded; this preserves the same published inventories and standalone Liquidity Score and V9 inputs without retaining a duplicate target graph at the scoring peak. Current EVM target/profile JSON is read through joined target-ID keyset pages of at most 32 rows, and each raw page is released before the next one is requested. The EVM scorer retains schema-validated profiles in serialized form, materializes proof graphs only for the target currently undergoing history and consumer validation, and attaches only proof-free public projections to retained pools. The 30-day confidence-history reader uses stablecoin/date keyset pages of at most 512 rows and releases every consumed page before requesting the next one, preserving the complete durability input without materializing the full history table beside the assembled pool graph. The main public-table mirror and generation-state transition remain one final two-statement D1 batch after exact candidate coverage validation, so partial staging never becomes current and failed publication does not advance freshness. Superseded, failed, and rejected measured-execution generations are retained for four hours — an hour above the three-hour freshness ceiling, so a profile can never read fresh after its backing rows were pruned — while completed `dex_liquidity_run_rows` generations no longer referenced by the public table and abandoned price run rows are retained for three hours; all are then pruned oldest-first in bounded producer-owned passes. Generation-ledger rows are removed only after their data rows are gone; measured published generations, active/incomplete work, the current liquidity generation, and any target generation still referenced by a retained quote remain protected regardless of age. Staged discovery pools retain 30 hours for the complete 24-hour scoring lookback, while provider `raw_json` is nulled after four hours; both passes are bounded to 1,000 oldest rows. Public `dex_liquidity_history` remains unchanged at 365 days. The score-bearing loader preserves the complete three-hour measured-execution history window while reading proof-heavy EVM history in sequential 16-target batches and releasing raw target, quote, and history rows as soon as each validated object is constructed. Raw producer envelopes (`raw_quote_payload_json`) are persisted only for failed quotes, where they are the sole structured failure evidence; measured quotes carry their complete evidence in the validated profile's `quoteProof`, and the score-bearing evidence loader does not select the raw column.
 
-Measured target inventories are published before proof-bearing quote evidence is loaded. After each EVM evidence family validates its targets, the scorer immediately releases that family's target descriptors, proof profiles, and internal diagnostics while retaining the proof-free public projection, physical-pool identity, and fail-closed gate consumed by P4. Evidence maps are then cleared before the next family loads. This bounds proof-heavy object lifetime without changing validation, activation policy, target coverage, or public score inputs.
+Measured target inventories are prepared before proof-bearing quote evidence is loaded, but remain private until the candidate passes all source, coverage, value, and major-asset guards and its full liquidity generation has persisted. A rejected candidate or failed liquidity write preserves the previously published active and shadow target catalogs. Only bounded target descriptors survive to this publication step; producer maps and quote proofs are still released during scoring. Odd-hour price-only passes do not publish target catalogs, and shadow targets retain their daily publication cadence. After each EVM evidence family validates its targets, the scorer immediately releases that family's target descriptors, proof profiles, and internal diagnostics while retaining the proof-free public projection, physical-pool identity, and fail-closed gate consumed by P4. Evidence maps are then cleared before the next family loads. This bounds proof-heavy object lifetime without changing validation, activation policy, target coverage, or public score inputs.
 
 Retained-route discovery applies its current-target, last-known-good, maturity, adapter, and tracked-asset checks from compact target/history metadata before parsing serialized EVM proof profiles. Only an absent eligible route materializes a full profile, and an accepted Curve packet reuses that parsed profile during packet validation. Current measured targets therefore do not pay a second proof-materialization pass solely to establish that no retained route is needed.
 

worker/src/cron/__tests__/dex-liquidity-scoring.test.ts

@@ -60,6 +60,7 @@ import {
   computeDepthStability,
   computeDexPrices,
   computeStablecoinScores,
+  publishStablecoinScoreTargets,
   DEX_LIQUIDITY_SCORING_BATCH_SIZE,
   loadConfidentHistoryStability,
   selectDexRouteObservations,
@@ -1353,6 +1354,14 @@ describe("dex-liquidity scoring", () => {
       shadowTargets,
     );
 
+    expect(result.diagnostics.measuredExecution.shadowTargetPublication).toEqual({
+      status: "skipped", reason: "liquidity-candidate-not-published",
+    });
+    expect(db.scoringTestState.measuredTargetRows).toBe(0);
+    expect(result.measuredTargetInventory.shadow).toHaveLength(1);
+    await publishStablecoinScoreTargets(db, result.measuredTargetInventory, result.diagnostics, 1_700_000_100);
+    expect(result.measuredTargetInventory.shadow).toHaveLength(0);
+
     expect(result.diagnostics.measuredExecution).toMatchObject({
       inventoryTargetCount: 0,
       shadowInventoryTargetCount: 1,

worker/src/cron/__tests__/sync-dex-liquidity.test.ts

@@ -54,6 +54,7 @@ const phaseFixtures = vi.hoisted(() => {
         priceObservations: new Map(),
       },
       scores: {
+        measuredTargetInventory: { mode: "active", active: [], shadow: [] },
         scores: new Map([["usdt-tether", { coverageClass: "primary", tvl: 0 }]]),
         globalAgg: { totalTvl: 0 },
         retainedPoolsByStablecoin: new Map(),
@@ -123,6 +124,7 @@ vi.mock("../dex-liquidity/staging-merge", () => ({
 vi.mock("../dex-liquidity/scoring", () => ({
   loadCurrentDexScoringGenerationId: vi.fn(async () => null),
   computeStablecoinScores: vi.fn(async () => phaseFixtures.current.scores),
+  publishStablecoinScoreTargets: vi.fn(async () => {}),
   computeDepthStability: vi.fn(async () => {}),
   computeDexPrices: vi.fn(async () => {}),
 }));
@@ -267,6 +269,7 @@ import {
   computeDepthStability,
   computeDexPrices,
   computeStablecoinScores,
+  publishStablecoinScoreTargets,
   loadCurrentDexScoringGenerationId,
 } from "../dex-liquidity/scoring";
 import { persistScores, writeHistoricalSnapshots } from "../dex-liquidity/persistence";
@@ -397,6 +400,7 @@ describe("dex liquidity scoring stage cycle", () => {
     expect(metadata.persistence?.skipped).toBe(true);
     expect(metadata.persistence?.skippedReason).toBe("defillama-protocols-unavailable");
     expect(persistScores).not.toHaveBeenCalled();
+    expect(publishStablecoinScoreTargets).not.toHaveBeenCalled();
     expect(computeDexPrices).not.toHaveBeenCalled();
     expect(writeHistoricalSnapshots).not.toHaveBeenCalled();
     expect(computeDepthStability).not.toHaveBeenCalled();
@@ -422,6 +426,7 @@ describe("dex liquidity scoring stage cycle", () => {
     };
     expect(metadata.failedSources).toContain("defillama-yields");
     expect(persistScores).not.toHaveBeenCalled();
+    expect(publishStablecoinScoreTargets).not.toHaveBeenCalled();
   });
 
   it("degrades and skips persistence instead of tripping hard value guard when DL yields is unavailable", async () => {
@@ -436,14 +441,15 @@ describe("dex liquidity scoring stage cycle", () => {
       dlProtocolsAvailable: true,
     });
     vi.mocked(computeStablecoinScores).mockResolvedValueOnce({
+      measuredTargetInventory: { mode: "active", active: [], shadow: [] },
       scores: new Map([["usdt-tether", { coverageClass: "primary", tvl: 2_000_000_000 }]]),
       globalAgg: { totalTvl: 2_000_000_000 },
       retainedPoolsByStablecoin: new Map(),
       tvlStabilityMap: new Map(),
       diagnostics: {
         protocolCapReductions: { cappedPoolCount: 0, cappedProtocols: 0, reducedTvlUsd: 0 },
       },
-    } as Awaited<ReturnType<typeof computeStablecoinScores>>);
+    } as unknown as Awaited<ReturnType<typeof computeStablecoinScores>>);
     const guardDb = makeNoopD1({
       prepare(sql: string) {
         if (sql.includes("COUNT(*) as cnt FROM dex_liquidity")) {
@@ -496,9 +502,48 @@ describe("dex liquidity scoring stage cycle", () => {
     expect(metadata.sourceCoverage?.currentGlobalTvl).toBe(2_000_000_000);
     expect(metadata.sourceCoverage?.nearValueGuard).toBe(true);
     expect(persistScores).not.toHaveBeenCalled();
+    expect(publishStablecoinScoreTargets).not.toHaveBeenCalled();
     expect(computeDexPrices).not.toHaveBeenCalled();
   });
 
+  it.each(["critical-source", "coverage", "value", "persistence"] as const)(
+    "retains the measured target catalog after %s rejection and publishes only after accepted liquidity",
+    async (rejection) => {
+      let catalog = "accepted-targets";
+      vi.mocked(publishStablecoinScoreTargets).mockImplementationOnce(async () => { catalog = "candidate-targets"; });
+      if (rejection === "critical-source") phaseFixtures.current.primary.dlYieldsAvailable = false;
+      if (rejection === "persistence") vi.mocked(persistScores).mockRejectedValueOnce(new Error("liquidity publication failed"));
+      const guardDb = makeNoopD1({
+        prepare(sql: string) {
+          if (rejection === "coverage" && sql.includes("COUNT(*) as cnt FROM dex_liquidity")) {
+            return { first: async () => ({ cnt: 100 }) };
+          }
+          if (rejection === "value" && sql.includes("SELECT total_tvl_usd, updated_at FROM dex_liquidity")) {
+            return { first: async () => ({ total_tvl_usd: 1_000_000_000, updated_at: 1_777_556_412 }) };
+          }
+          return db.prepare(sql);
+        },
+      });
+      if (rejection === "critical-source") {
+        expect((await runDexLiquidityScoringCycle(guardDb, "graph-key")).status).toBe("degraded");
+      } else {
+        await expect(runDexLiquidityScoringCycle(guardDb, "graph-key")).rejects.toThrow(
+          rejection === "persistence" ? "liquidity publication failed" : "coverage guard tripped",
+        );
+      }
+      expect(catalog).toBe("accepted-targets");
+      expect(publishStablecoinScoreTargets).not.toHaveBeenCalled();
+
+      phaseFixtures.reset();
+      expect((await runDexLiquidityScoringCycle(db, "graph-key")).status).toBe("ok");
+      expect(catalog).toBe("candidate-targets");
+      expect(publishStablecoinScoreTargets).toHaveBeenCalledOnce();
+      expect(vi.mocked(publishStablecoinScoreTargets).mock.invocationCallOrder[0]).toBeGreaterThan(
+        vi.mocked(persistScores).mock.invocationCallOrder[vi.mocked(persistScores).mock.invocationCallOrder.length - 1]!,
+      );
+    },
+  );
+
   it("returns ok when required source families succeed", async () => {
     const result = await runDexLiquidityScoringCycle(db, "graph-key");
 
@@ -581,6 +626,7 @@ describe("dex liquidity scoring stage cycle", () => {
     const scoreCalls = vi.mocked(computeStablecoinScores).mock.calls;
     expect(scoreCalls[scoreCalls.length - 1]?.[8]).toBe("none");
     expect(persistScores).not.toHaveBeenCalled();
+    expect(publishStablecoinScoreTargets).not.toHaveBeenCalled();
     expect(writeHistoricalSnapshots).not.toHaveBeenCalled();
     expect(computeDepthStability).not.toHaveBeenCalled();
     expect(computeDexPrices).toHaveBeenCalledOnce();
@@ -1190,6 +1236,7 @@ describe("dex liquidity scoring stage cycle", () => {
       ]),
     );
     vi.mocked(computeStablecoinScores).mockResolvedValueOnce({
+      measuredTargetInventory: { mode: "active", active: [], shadow: [] },
       scores: new Map([
         [
           "usdc-circle",
@@ -1220,7 +1267,7 @@ describe("dex liquidity scoring stage cycle", () => {
       diagnostics: {
         protocolCapReductions: { cappedPoolCount: 1, cappedProtocols: 1, reducedTvlUsd: 50 },
       },
-    } as Awaited<ReturnType<typeof computeStablecoinScores>>);
+    } as unknown as Awaited<ReturnType<typeof computeStablecoinScores>>);
 
     const driftDb = makeNoopD1({
       prepare(sql: string) {

worker/src/cron/dex-liquidity/orchestrator.ts

@@ -24,6 +24,7 @@ import {
 import { mergeStagedPools } from "./staging-merge";
 import {
   computeStablecoinScores,
+  publishStablecoinScoreTargets,
   computeDepthStability,
   computeDexPrices,
   loadCurrentDexScoringGenerationId,
@@ -423,6 +424,7 @@ interface DexLiquidityScoreState {
   retainedPoolsByStablecoin: Map<string, LiquidityMetrics["topPools"]>;
   tvlStabilityMap: Map<string, number>;
   diagnostics: Awaited<ReturnType<typeof computeStablecoinScores>>["diagnostics"];
+  measuredTargetInventory: Awaited<ReturnType<typeof computeStablecoinScores>>["measuredTargetInventory"];
   analysis: DexLiquidityAnalysis;
 }
 
@@ -917,6 +919,7 @@ async function scoreDexLiquidityPoolState(
     globalAgg,
     retainedPoolsByStablecoin,
     tvlStabilityMap,
+    measuredTargetInventory,
     diagnostics,
   } = await computeStablecoinScores(
     ctx.db,
@@ -990,6 +993,7 @@ async function scoreDexLiquidityPoolState(
     globalAgg,
     retainedPoolsByStablecoin,
     tvlStabilityMap,
+    measuredTargetInventory,
     diagnostics,
     analysis,
   };
@@ -1004,6 +1008,8 @@ async function persistDexLiquidityScoreState(
 ): Promise<DexLiquidityPersistenceState> {
   const skippedReason = getPersistenceSkipReason(sourceState.criticalSourceFailures);
   if (skippedReason) {
+    scoreState.measuredTargetInventory.active.length = 0;
+    scoreState.measuredTargetInventory.shadow.length = 0;
     await ctx.reportDexProgress("persistence-skipped", {
       message: `Skipping DEX liquidity publication: ${skippedReason}`, providerFamily: "d1", total: scoreState.scoreResults.size,
       metadata: { skippedReason, failedSources: sourceState.failedSources }, counts: { candidateRows: scoreState.scoreResults.size },
@@ -1077,6 +1083,15 @@ async function persistDexLiquidityScoreState(
     throw new Error("DEX liquidity persistence completed without a publication generation id");
   }
   poolState.metrics.clear();
+  if (options.publishLiquidity) {
+    await publishStablecoinScoreTargets(
+      ctx.db,
+      scoreState.measuredTargetInventory,
+      scoreState.diagnostics,
+      ctx.syncStartSec,
+      ctx.signal,
+    );
+  }
   await ctx.reportDexProgress(
     options.publishLiquidity ? "persistence-generation-complete" : "persistence-generation-reused",
     {