fix(news): unify digest publication acceptance (#7599)

* fix(news): unify digest publication acceptance

Keep canonical and durable digest snapshots behind one acceptance gate so a narrower candidate cannot land first.

* fix(news): close digest publication review gaps

* fix(news): avoid false digest failure after build

Author: koala73

docker/redis-rest-proxy.mjs

@@ -118,6 +118,27 @@ const DIGEST_LASTGOOD_PUBLISH_SCRIPT = [
   'local candidate = nil',
   'if okCandidate then candidate = countData(candidateData) end',
   'if not candidate or candidate.categories < 1 or candidate.items < 1 then return -1 end',
+  'local canonicalRaw = nil',
+  "if KEYS[3] then canonicalRaw = redis.call('GET', KEYS[3]) end",
+  'local function rejectNarrower()',
+  "  if KEYS[3] and not canonicalRaw then redis.call('SET', KEYS[3], '\"__WM_NEG__\"', 'EX', ARGV[9]) end",
+  '  return 0',
+  'end',
+  'local function isNarrower(nextData, currentData)',
+  '  return nextData.categories < currentData.categories or nextData.items < currentData.items',
+  'end',
+  'local function isLiveCanonicalClock(value)',
+  "  if type(value) ~= 'string' then return false end",
+  "  local year, month, day, hour, minute, second = string.match(value, '^(%d%d%d%d)%-(%d%d)%-(%d%d)T(%d%d):(%d%d):(%d%d)%.%d%d%dZ$')",
+  '  if not year then return false end',
+  '  year, month, day = tonumber(year), tonumber(month), tonumber(day)',
+  '  hour, minute, second = tonumber(hour), tonumber(minute), tonumber(second)',
+  '  if month < 1 or month > 12 or hour > 23 or minute > 59 or second > 59 then return false end',
+  '  local leap = year % 4 == 0 and (year % 100 ~= 0 or year % 400 == 0)',
+  '  local monthDays = { 31, leap and 29 or 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31 }',
+  '  if day < 1 or day > monthDays[month] then return false end',
+  '  return value >= ARGV[7] and value <= ARGV[8]',
+  'end',
   "local currentRaw = redis.call('GET', KEYS[1])",
   'if currentRaw then',
   '  local okCurrent, snapshot = pcall(cjson.decode, currentRaw)',
@@ -126,7 +147,18 @@ const DIGEST_LASTGOOD_PUBLISH_SCRIPT = [
   '    if current then',
   '      local delta = tonumber(ARGV[1]) - (tonumber(snapshot.acceptedAt) or 0)',
   '      local live = delta >= 0 and delta <= tonumber(ARGV[2])',
-  '      if live and (candidate.categories < current.categories or candidate.items < current.items) then return 0 end',
+  '      if live and isNarrower(candidate, current) then return rejectNarrower() end',
+  '    end',
+  '  end',
+  'end',
+  'if KEYS[3] then',
+  '  if canonicalRaw then',
+  '    local okCanonical, canonicalData = pcall(cjson.decode, canonicalRaw)',
+  "    if okCanonical and type(canonicalData) == 'table' then",
+  '      local currentCanonical = countData(canonicalData)',
+  '      local live = isLiveCanonicalClock(canonicalData.generatedAt)',
+  '      local usable = currentCanonical and currentCanonical.categories >= 1 and currentCanonical.items >= 1',
+  '      if live and usable and isNarrower(candidate, currentCanonical) then return rejectNarrower() end',
   '    end',
   '  end',
   'end',
@@ -140,6 +172,7 @@ const DIGEST_LASTGOOD_PUBLISH_SCRIPT = [
   "  .. ',\"itemCount\":' .. string.format('%.0f', candidate.items)",
   '  .. \',"data":\' .. ARGV[5] .. \'}\'',
   "redis.call('SET', KEYS[1], stored, 'EX', ARGV[4])",
+  "if KEYS[3] then redis.call('SET', KEYS[3], ARGV[5], 'EX', ARGV[6]) end",
   'return 1',
 ].join('\n');
 

server/_shared/redis.ts

@@ -252,8 +252,7 @@ export async function getCachedEnvelopeJson(key: string, raw = false): Promise<u
 export async function setCachedJson(key: string, value: unknown, ttlSeconds: number, raw = false): Promise<boolean> {
   if (process.env.LOCAL_API_MODE === 'tauri-sidecar') {
     const { sidecarCacheSet } = await import('./sidecar-cache');
-    sidecarCacheSet(key, value, ttlSeconds);
-    return true;
+    return sidecarCacheSet(key, value, ttlSeconds);
   }
 
   const url = process.env.UPSTASH_REDIS_REST_URL;
@@ -448,8 +447,7 @@ export async function prependCachedJsonList(
       const retained = Array.isArray(existing)
         ? existing.filter((item) => JSON.stringify(item) !== encoded)
         : [];
-      sidecarCacheSet(key, [value, ...retained].slice(0, boundedLimit), boundedTtlSeconds);
-      return true;
+      return sidecarCacheSet(key, [value, ...retained].slice(0, boundedLimit), boundedTtlSeconds);
     } catch (err) {
       // sentry-coverage-ok: this helper returns false to its caller, and a
       // history write must never fail the current response it will inform.
@@ -987,15 +985,19 @@ export interface UsageHook {
  * `opts.cacheFailures: false` similarly makes nulls, no-store payloads, and
  * thrown fetches non-cacheable. `opts.inflightKey` lets callers share positive
  * cache entries while isolating provider-local work and failures.
+ * `opts.onPositiveResult` keeps the in-flight slot open for a caller-owned
+ * commit phase after a valid fetch result.
  * `opts.isCallerLocalError` identifies admission failures that belong only to
  * the fetch leader. These errors do not arm shared cache backoff, and an
  * in-flight follower re-enters admission under its own request instead of
  * inheriting any preceding leader's failure.
  */
-type CachedFetchWithMetaOpts = CachedFetchOpts & {
+type CachedFetchWithMetaOpts<T extends object = object> = CachedFetchOpts & {
   usage?: UsageHook;
   shouldFetch?: () => boolean;
   cacheFailures?: boolean;
+  cachePositiveResult?: boolean;
+  onPositiveResult?: (result: T) => Promise<void>;
   inflightKey?: string;
   isCallerLocalError?: (error: unknown) => boolean;
 };
@@ -1005,7 +1007,7 @@ export async function cachedFetchJsonWithMeta<T extends object>(
   ttlSeconds: number,
   fetcher: () => Promise<T | null>,
   negativeTtlSeconds = 120,
-  opts?: CachedFetchWithMetaOpts,
+  opts?: CachedFetchWithMetaOpts<T>,
 ): Promise<{ data: T | null; source: 'cache' | 'fresh' | 'skipped'; leader: boolean }> {
   return cachedFetchJsonCore(key, ttlSeconds, fetcher, negativeTtlSeconds, opts, 'cachedFetchJsonWithMeta');
 }
@@ -1023,7 +1025,7 @@ async function cachedFetchJsonCore<T extends object>(
   ttlSeconds: number,
   fetcher: () => Promise<T | null>,
   negativeTtlSeconds: number,
-  opts: CachedFetchWithMetaOpts | undefined,
+  opts: CachedFetchWithMetaOpts<T> | undefined,
   callerName: 'cachedFetchJson' | 'cachedFetchJsonWithMeta',
 ): Promise<{ data: T | null; source: 'cache' | 'fresh' | 'skipped'; leader: boolean }> {
   const cached = await readCachedJson(key);
@@ -1093,12 +1095,16 @@ async function cachedFetchJsonCore<T extends object>(
           }
         } else {
           upstreamStatus = 200;
-          const wrote = await setCachedJson(key, result, ttlSeconds);
+          const wrote = opts?.cachePositiveResult === false
+            ? true
+            : await setCachedJson(key, result, ttlSeconds);
           // See cachedFetchJson(): this short in-process bridge is only for
           // remote Redis outages, not local sidecar cache writes.
-          if (hadCacheReadError || (!wrote && hasRemoteRedisConfig())) {
+          if (opts?.cachePositiveResult !== false
+            && (hadCacheReadError || (!wrote && hasRemoteRedisConfig()))) {
             armLocalPositiveFallback(key, result, ttlSeconds);
           }
+          if (opts?.cachePositiveResult === false) await opts.onPositiveResult?.(result);
         }
       } else {
         upstreamStatus = 0;

server/_shared/sidecar-cache.ts

@@ -77,11 +77,10 @@ export function sidecarCacheGet(key: string): unknown | null {
 export function sidecarCacheSetIfAbsent(key: string, value: unknown, ttlSeconds: number): boolean {
   const existing = store.get(key);
   if (existing && existing.expiresAt > Date.now()) return false;
-  sidecarCacheSet(key, value, ttlSeconds);
-  return true;
+  return sidecarCacheSet(key, value, ttlSeconds);
 }
 
-export function sidecarCacheSet(key: string, value: unknown, ttlSeconds: number): void {
+export function sidecarCacheSet(key: string, value: unknown, ttlSeconds: number): boolean {
   const clamped = Math.max(MIN_TTL_S, Math.min(MAX_TTL_S, ttlSeconds));
   const json = JSON.stringify(value);
   // Rough byte estimate: JS strings are UTF-16 (2 bytes per code unit).
@@ -90,7 +89,7 @@ export function sidecarCacheSet(key: string, value: unknown, ttlSeconds: number)
 
   if (size > MAX_SINGLE_VALUE_BYTES) {
     console.warn(`[sidecar-cache] rejecting key "${key}": ${(size / 1024 / 1024).toFixed(1)} MB exceeds 2 MB limit`);
-    return;
+    return false;
   }
 
   // Remove old entry if exists
@@ -113,6 +112,7 @@ export function sidecarCacheSet(key: string, value: unknown, ttlSeconds: number)
   totalBytes += size;
 
   startSweepIfNeeded();
+  return true;
 }
 
 export function sidecarCacheStats(): { entries: number; bytes: number; hits: number; misses: number } {

server/worldmonitor/news/v1/_lastgood-store.ts

@@ -15,7 +15,7 @@ import {
   runRedisTransaction,
   setCachedJson,
 } from '../../../_shared/redis';
-import { REVOKED_URLS_KEY } from '../../../_shared/digest-revocations';
+import { REVOKED_URLS_KEY, readRevokedUrlSet } from '../../../_shared/digest-revocations';
 import { getUsageScope } from '../../../_shared/usage';
 import {
   ATTEMPT_META_TTL_S,
@@ -36,6 +36,8 @@ import {
 // Private wire value used by cachedFetchJsonWithMeta. This write is kept here
 // because the attempt row and sentinel must become visible atomically.
 const DIGEST_NEGATIVE_SENTINEL = '__WM_NEG__';
+const DIGEST_CACHE_TTL_S = 900;
+const DIGEST_REJECTION_TTL_S = 120;
 
 export interface FailedDigestAttempt {
   readonly at: string;
@@ -87,6 +89,11 @@ export function shouldStartDigestAttempt(digestCacheKey: string, now = Date.now(
   return false;
 }
 
+export function deferDigestAttempt(digestCacheKey: string, ttlSeconds: number, now = Date.now()): void {
+  boundLocalRecoveryMaps(now);
+  failureCooldowns.set(digestCacheKey, now + ttlSeconds * 1000);
+}
+
 export function beginDigestAttempt(variant: string, lang: string, at: string): AttemptSlot {
   // A shared promise rejection can wake a follower before the leader's outer
   // catch runs. The thrown/timeout identity is therefore prepared at start;
@@ -120,7 +127,8 @@ export function measureServableRichness(
   const categories = Object.values(data.categories ?? {});
   let itemCount = 0;
   for (const bucket of categories) {
-    for (const item of bucket?.items ?? []) {
+    const items = Array.isArray(bucket?.items) ? bucket.items : [];
+    for (const item of items) {
       const link = item && typeof item === 'object' && 'link' in item
         ? (item as { link?: unknown }).link
         : undefined;
@@ -262,60 +270,145 @@ export async function publishAcceptedSnapshot(
   variant: string,
   lang: string,
   data: ListFeedDigestResponse,
-): Promise<void> {
-  if (!isEligibleScope(variant, lang) || !isAcceptableDigest(data)) return;
+  canonicalDigestKey?: string,
+): Promise<'accepted' | 'rejected' | 'unavailable'> {
+  if (!isEligibleScope(variant, lang) || !isAcceptableDigest(data)) return 'rejected';
   const now = Date.now();
   const generatedAtMs = Date.parse(data.generatedAt ?? '');
   const acceptedAt = Number.isFinite(generatedAtMs) ? generatedAtMs : now;
   try {
     if (process.env.LOCAL_API_MODE === 'tauri-sidecar') {
+      const revoked = await readRevokedUrlSet();
+      if (!revoked.readable) {
+        console.warn(`[digest-publication] source unavailable variant=${variant} lang=${lang}`);
+        return 'unavailable';
+      }
+      const candidateRichness = measureServableRichness(data, revoked.urls);
+      if (candidateRichness.categoryCount < 1 || candidateRichness.itemCount < 1) {
+        console.log(`[digest-publication] candidate rejected after revocations variant=${variant} lang=${lang}`);
+        return 'rejected';
+      }
       const read = await readAcceptedSnapshot<ListFeedDigestResponse>(variant, lang);
-      if (!read.readable) return;
+      if (!read.readable) {
+        console.warn(`[digest-publication] publish unavailable variant=${variant} lang=${lang}`);
+        return 'unavailable';
+      }
       const current = read.snapshot;
-      const { categoryCount, itemCount } = measureServableRichness(data, new Set());
-      const decision = shouldReplaceAccepted(current, { categoryCount, itemCount }, now);
-      if (!decision.replace) return;
-      const meta: AcceptedSnapshotMeta = { acceptedAt, categoryCount, itemCount };
-      await setCachedJson(lastGoodKey(variant, lang), { ...meta, data }, LASTGOOD_TTL_S);
-      return;
+      const currentCanonical = canonicalDigestKey
+        ? await readCachedJson(canonicalDigestKey)
+        : { status: 'miss' as const };
+      if (currentCanonical.status === 'error') {
+        console.warn(`[digest-publication] canonical read unavailable variant=${variant} lang=${lang}`);
+        return 'unavailable';
+      }
+      const canonicalValue = currentCanonical.status === 'hit'
+        && currentCanonical.value && typeof currentCanonical.value === 'object'
+        ? currentCanonical.value as ListFeedDigestResponse
+        : null;
+      const canonicalGeneratedAt = canonicalValue ? Date.parse(canonicalValue.generatedAt ?? '') : NaN;
+      const canonicalRichness = canonicalValue
+        ? measureServableRichness(canonicalValue, revoked.urls)
+        : null;
+      const canonicalMeta = canonicalRichness
+        && canonicalRichness.categoryCount >= 1
+        && canonicalRichness.itemCount >= 1
+        && Number.isFinite(canonicalGeneratedAt)
+        ? {
+            acceptedAt: canonicalGeneratedAt,
+            ...canonicalRichness,
+          }
+        : null;
+      const decision = shouldReplaceAccepted(current, candidateRichness, now);
+      const canonicalDecision = canonicalMeta
+        ? shouldReplaceAccepted(canonicalMeta, candidateRichness, now)
+        : null;
+      if (!decision.replace || canonicalDecision && !canonicalDecision.replace) {
+        if (!decision.replace && canonicalDigestKey && currentCanonical.status === 'miss') {
+          const cooldownWritten = await setCachedJson(
+            canonicalDigestKey,
+            DIGEST_NEGATIVE_SENTINEL,
+            DIGEST_REJECTION_TTL_S,
+          );
+          if (!cooldownWritten) {
+            console.warn(`[digest-publication] publish unavailable variant=${variant} lang=${lang}`);
+            return 'unavailable';
+          }
+        }
+        console.log(`[digest-publication] candidate rejected by acceptance gate variant=${variant} lang=${lang}`);
+        return 'rejected';
+      }
+      const meta: AcceptedSnapshotMeta = { acceptedAt, ...candidateRichness };
+      const durableWritten = await setCachedJson(lastGoodKey(variant, lang), { ...meta, data }, LASTGOOD_TTL_S);
+      if (!durableWritten) {
+        console.warn(`[digest-publication] publish unavailable variant=${variant} lang=${lang}`);
+        return 'unavailable';
+      }
+      if (canonicalDigestKey) {
+        const canonicalWritten = await setCachedJson(canonicalDigestKey, data, DIGEST_CACHE_TTL_S);
+        if (!canonicalWritten) {
+          console.warn(`[digest-publication] publish unavailable variant=${variant} lang=${lang}`);
+          return 'unavailable';
+        }
+      }
+      return 'accepted';
     }
 
     // ARGV[5] is the digest body ALONE, and the script splices it into the
     // stored JSON verbatim. Sending the wrapped `{ acceptedAt, data }` and
     // letting Lua rebuild it meant a cjson decode/encode round trip, which
     // silently rewrote every empty array in the body as `{}`.
-    const results = await runRedisPipeline([[
-      'EVAL',
-      DIGEST_LASTGOOD_PUBLISH_SCRIPT,
-      '2',
-      lastGoodKey(variant, lang),
-      REVOKED_URLS_KEY,
+    const keys = canonicalDigestKey
+      ? [lastGoodKey(variant, lang), REVOKED_URLS_KEY, canonicalDigestKey]
+      : [lastGoodKey(variant, lang), REVOKED_URLS_KEY];
+    const args = [
       String(now),
       String(LASTGOOD_MAX_AGE_MS),
       String(acceptedAt),
       String(LASTGOOD_TTL_S),
       JSON.stringify(data),
+      ...(canonicalDigestKey ? [
+        String(DIGEST_CACHE_TTL_S),
+        new Date(now - LASTGOOD_MAX_AGE_MS).toISOString(),
+        new Date(now).toISOString(),
+        String(DIGEST_REJECTION_TTL_S),
+      ] : []),
+    ];
+    const results = await runRedisPipeline([[
+      'EVAL',
+      DIGEST_LASTGOOD_PUBLISH_SCRIPT,
+      String(keys.length),
+      ...keys,
+      ...args,
     ]]);
     const outcome = results[0];
     if (!outcome || outcome.error) {
-      console.warn(`[digest-lastgood] guarded publish unavailable variant=${variant} lang=${lang}`);
+      console.warn(`[digest-publication] publish unavailable variant=${variant} lang=${lang}`);
+      return 'unavailable';
     } else if (outcome.result === 0) {
-      console.log(`[digest-lastgood] kept live snapshot (not-narrower) variant=${variant} lang=${lang}`);
+      console.log(`[digest-publication] candidate rejected by acceptance gate variant=${variant} lang=${lang}`);
+      return 'rejected';
     } else if (outcome.result === -1) {
-      console.log(`[digest-lastgood] candidate rejected after revocations variant=${variant} lang=${lang}`);
+      console.log(`[digest-publication] candidate rejected after revocations variant=${variant} lang=${lang}`);
+      return 'rejected';
+    } else if (outcome.result === 1) {
+      return 'accepted';
     }
+    console.warn(`[digest-publication] publish unavailable variant=${variant} lang=${lang}`);
+    return 'unavailable';
   } catch (err) {
-    console.warn('[digest-lastgood] publish failed:', err);
+    console.warn('[digest-publication] publish failed:', err);
     captureSilentError(err, {
       tags: { surface: 'news', component: 'digest-lastgood', stage: 'publish', variant, lang },
       fingerprint: ['digest-lastgood', 'publish-failed'],
     });
+    return 'unavailable';
   }
 }
 
 export const __testing__ = {
   activeAttempts,
   recentFailedAttempts,
   failureCooldowns,
+  deferDigestAttempt,
   measureServableRichness,
 };