ROX-37813: Add write-time compliance freshness hooks - #23765

Populate the data_freshness column through two write-time events, gated
on the ComplianceSurfaceStaleData feature flag:

- CURRENT on receive: the results pipeline marks a check result CURRENT
  right before upsert, so new data reads as fresh immediately.
- Prior-only STALE on timeout: when a scan times out, the report manager
  marks STALE only the results the timed-out run did NOT refresh
  (LastStartedTime older than the run's start time). Results the run did
  refresh are genuinely fresh and are left unchanged. Marking the whole
  scan's results stale was wrong (the same row is reused across runs, so
  it would flag fresh overwrites as stale).

Both write paths mutate freshness inside the serialized proto, so they
do a read-modify-upsert. A concurrent ingest and the timeout hook could
otherwise revert each other (lost update). A per-row KeyFence on the
check-result IDs serializes the ingest upsert against the timeout hook's
read-modify-upsert, mirroring the CVE suppress precedent.

Scope note: without a background sweeper (next slice), data cannot decay
from CURRENT to STALE except through an explicit event (a scan timeout).
The silent-decay case (schedule stalls, sensor permanently disconnected,
no event ever fires) is not covered until the sweeper lands.

Considered alternative for the timeout hook: mark all existing results
stale at scan start and let arrivals flip them back to CURRENT. Rejected
because it flaps freshness on every normal scan and needs reliable
scan-start detection. The chosen approach runs only on a real timeout
and reuses a signal already in hand (the run's start time).

Tests: unit (CURRENT-on-receive pipeline test, timeout mark-stale
manager test) and sql_integration (prior-only STALE, plus a concurrency
regression test proving the KeyFence stops the ingest and timeout hook
from reverting each other - verified to fail without the fence).

This code was partially generated by AI (opencode).

Review follow-up: "received" does not always mean "new". Sensor
resends every existing CheckResult CR on reconnect regardless of
whether anything changed (compliance resources are not deduplicated),
and a misbehaving compliance operator could republish a result without
a real rescan behind it. In both cases LastStartedTime (parsed from
the operator's own last-scanned-timestamp annotation) is unchanged, so
marking it CURRENT unconditionally would launder stale data as fresh.
There is no sweeper in this slice to correct that later, unlike the
original PoC design, so the gap had to be closed here.

Before marking CURRENT, compare the incoming LastStartedTime against
the currently stored row (if any): only a strictly newer timestamp (or
no prior row at all) counts as a genuinely new scan result. Otherwise
the row's existing freshness is carried forward unchanged, since the
freshly built proto would otherwise default to UNKNOWN and wipe out
the prior value on upsert.

Added a regression test for the replay case (verified to fail without
the fix: wants the stale row's freshness preserved, previously got
CURRENT), plus a test for the newer-than-stored case.

This code was partially generated by AI (opencode).

Review findings (CodeRabbit + an independent adversarial review)
on #23765: the freshness-on-receive comparison read the existing row
via GetComplianceCheckResult outside the KeyFence, then wrote the
decision later through UpsertResult inside the fence. Two problems
followed from that gap between read and write:

1. A replayed result (same LastStartedTime, so not genuinely new)
   could be decided from a read that predates a concurrent timeout
   marking, then overwrite that marking's STALE value when it finally
   wrote -- the lock only protected the write, not the read-decide-write
   sequence as a whole.
2. On a lookup error, the code fell back to UNKNOWN and still
   proceeded to upsert, silently replacing a known CURRENT/STALE value
   with UNKNOWN instead of aborting.

Fix: move the comparison into the datastore, behind a new
UpsertReceivedResult method that reads the existing row, derives
data_freshness, and upserts -- all inside the same per-row KeyFence
critical section already used by MarkStalePriorResultsByScanRef. The
two can no longer interleave: whichever one's critical section runs
second sees the other's result, so a replay converges on the
timeout's STALE value instead of reverting it. A lookup error now
aborts before any write.

The pipeline's role shrinks to routing: call UpsertReceivedResult when
the flag is on, plain UpsertResult otherwise. Its freshness-derivation
tests move to the datastore level, where the logic now lives; added a
regression test that races UpsertReceivedResult (a replay) against
MarkStalePriorResultsByScanRef (a timeout) and asserts the result
converges to STALE regardless of interleaving -- verified to fail
deterministically (5/5) against the prior unfenced-read code, passes
with the fix. Also added a start barrier to the existing concurrency
regression test so both sides are guaranteed to be running before
either does real work, instead of relying on incidental scheduling
overlap.

Two other findings from the adversarial review were checked and don't
hold: the scan's LastStartedTime used as the timeout pivot cannot be
nil in practice (GetWatcherIDFromScan rejects a scan with a nil start
time before a watcher is ever created for it, so handleReadyScan never
sees one), and the feature-flag env var set via T().Setenv() is read
live by features.X.Enabled() (confirmed: none of the flag-gated tests
skip).

This code was partially generated by AI (opencode), including review
by independent AI agents (CodeRabbit, and an adversarial Opus-model
review) whose findings drove this fix.

CI's golangci-lint flagged both classic index loops I added
(TestMarkStalePriorDoesNotRevertConcurrentUpsert,
TestUpsertReceivedResultDoesNotRevertConcurrentTimeout) with
rangeint/modernize: Go 1.26 supports range-over-int, so
'for i := 0; i < n; i++' becomes 'for i := range n'. No behavior
change, confirmed by re-running both tests.

This code was partially generated by AI (opencode).

Further CodeRabbit findings on #23765:

1. (Major) UpsertReceivedResult preserved DataFreshness on a replay but
   still upserted the whole incoming object, including its own
   LastStartedTime. An out-of-order or duplicate delivery carrying an
   OLDER LastStartedTime than what is stored would regress the stored
   timestamp, and a later timeout's age-based STALE decision
   (LastStartedTime < pivot) could then wrongly flag a genuinely
   fresh row stale. Fixed by folding the LastStartedTime decision into
   the same reconciliation as DataFreshness (renamed freshnessOnReceive
   to reconcileReceivedResult, now mutating both fields together under
   one comparison): a replay or older/nil incoming timestamp clamps
   LastStartedTime to the stored value instead of overwriting it.
   Verified the regression test fails without the fix (stored timestamp
   regressed) and passes with it.

2. (Minor) Flagged the !found-and-nil-timestamp branch returning
   UNKNOWN instead of CURRENT, citing UpsertReceivedResult's own doc
   comment ("CURRENT when it has no prior stored row"). Evaluated and
   NOT changed: with no sweeper in this slice, a wrongly-asserted
   CURRENT for a row with no timestamp would have no way back, because
   MarkStalePriorResultsByScanRef explicitly leaves nil-timestamp rows
   unchanged forever. UNKNOWN is the safer choice and self-corrects on
   the next update that does carry a timestamp. Fixed the actual
   inconsistency in the doc comment instead (datastore.go), and added
   a regression test for this exact branch.

3. (Minor) A watcher created from a check result alone (before its
   scan update ever arrives, see GetWatcherIDFromCheckResult) can time
   out with Scan never set. The timeout hook's epoch-pivot and
   empty-scan-ref calculation from a nil Scan happened to no-op through
   MarkStalePriorResultsByScanRef's existing empty-ref guard, but
   silently -- a future change to that guard could turn a nil Scan
   into real misbehavior. Added an explicit guard with a debug log, and
   a regression test asserting MarkStalePriorResultsByScanRef is never
   called in this case. Verified it fails without the guard
   (unexpected call with the 1970 epoch pivot and empty scan ref).

Also fixed a related, still-open finding from an earlier review round
on the same PR (MarkStalePriorResultsByScanRef's fenced re-read did
not re-check ScanRefId, so a check result whose scan-name label changed
between the initial candidate scan and the fenced re-read could have
its STALE decision wrongly driven by a different scan's pivot): added
a scanRefId equality check inside the want callback.

Added TestUpsertReceivedResult (sql_integration), a non-concurrent
test exercising all of reconcileReceivedResult's branches directly at
the datastore level, closing a coverage gap left when the logic moved
out of the (previously mock-tested) pipeline layer in 732480e.

This code was partially generated by AI (opencode), addressing further
CodeRabbit review findings on the same PR.

… helper to markStale

Remove the separate UpsertReceivedResult interface method and consolidate
freshness derivation into UpsertResult (gated by feature flag). Rename the
private helper updateFreshness → markStale and change its callback signature
from returning a DataFreshness enum to a bool predicate shouldStale. This
makes the helper's purpose explicit: it only flips rows to STALE, never to
any other state. The generic-over-any-freshness-value signature was
misleading since the helper was only ever used for STALE marking.

Also simplify reconcileReceivedResult to only preserve DataFreshness from
the existing row on replays, removing the LastStartedTime clamping logic
(timestamp regression on out-of-order replays is now accepted as documented
in test comments - a rare case corrected by the next scan).

Originating request: "Refactor the private freshness helper so it only marks
rows STALE (rename updateFreshness -> markStale, change the callback to a
bool predicate), since it only ever flips rows to STALE and the
generic-over-any-value signature was misleading."

Code partially generated by AI.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

The timeout branch read the staleness scope (scan ref + pivot start time)
from the watcher-recorded fields, then fell back to the Scan object. The
watcher's recordScope always populates those fields before any timeout, so
the Scan fallback was reachable only from tests that built ScanWatcherResults
literals without a recorded scope. Drop it; the existing nil-scope skip still
guards the impossible-in-production empty case.

Also rename the log-only scanName variable to displayID (it falls back to the
watcher ID, which is not a scan name) and trim the now-inaccurate comment.

Request: "remove the test-only Scan fallback, clarify scanName=WatcherID, make
the timeout branch clearer and trim comments."

Partially AI-generated.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>