Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions src/env.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -621,6 +621,20 @@ declare global {
* updates this var at the next rotation rather than needing a code change. See
* review/ledger-anchor-rekor.ts. */
LOOPOVER_LEDGER_ANCHOR_REKOR_SHARD_URL?: string;
/** External ledger anchoring (#9273/#9274, epic #9267): the target repo/branch/path for the git-commit
* anchoring backend. Owner and repo are BOTH required for git anchoring to run at all -- unset means
* that backend is simply not configured yet (Rekor still runs on its own), the same honest-degrade
* posture as an unset signing key. Branch defaults to "main", path defaults to "anchors.jsonl" if unset.
* See review/ledger-anchor-scheduler.ts's `resolveGitAnchorTarget`. */
LOOPOVER_LEDGER_ANCHOR_GIT_OWNER?: string;
LOOPOVER_LEDGER_ANCHOR_GIT_REPO?: string;
LOOPOVER_LEDGER_ANCHOR_GIT_BRANCH?: string;
LOOPOVER_LEDGER_ANCHOR_GIT_PATH?: string;
/** External ledger anchoring (#9274, epic #9267): the GitHub App installation id used to mint the token
* the git-commit backend commits with (via the same makeInstallationOctokit/withInstallationTokenRetry
* chokepoint every other GitHub write in this engine goes through). Unset (alongside the owner/repo
* pair above) means the git backend does not run this tick; Rekor is unaffected. */
LOOPOVER_LEDGER_ANCHOR_GIT_INSTALLATION_ID?: string;
/** Convergence (port): public OAuth draft-submission flow ported from reviewbot. When truthy, the
* /v1/drafts endpoints accept a contributor draft -> GitHub OAuth -> fork PR against the content repo.
* Default OFF — unset/false makes every draft endpoint 404 and writes nothing (byte-identical worker). */
Expand Down
6 changes: 6 additions & 0 deletions src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,12 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
// budget is reserved for webhooks (which drive timely reviews) instead of compounding the backlog; the next
// tick (~2 min) retries, and after the bucket resets the sweep resumes. Webhooks never pre-yield.
const jobs: JobMessage[] = [];
// #9274 (epic #9267): enqueued EVERY tick, unconditionally, rather than gated behind isHourly here. The
// seq-threshold trigger (>=256 records since the last anchor) must fire "independent of the hourly clock"
// per the issue's own requirement -- gating the enqueue itself to isHourly would silently make that
// impossible. The job's own decideLedgerAnchorSchedule (in ledger-anchor-scheduler.ts) is what actually
// decides whether THIS tick does anything; a cheap two-query no-op the overwhelming majority of ticks.
jobs.push({ type: "anchor-decision-ledger", requestedBy: "schedule", isHourly });
const selfHostedReviews = isSelfHostedReviewRuntime(env);
const queueSnapshot = selfHostedReviews
? await queueSnapshotFromBinding(env.JOBS).catch((error) => {
Expand Down
21 changes: 21 additions & 0 deletions src/queue/job-dispatch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,10 @@ import {
setAprRepoDispatchPaused,
} from "../orb/apr-repo-transfer";
import { syncBrokeredInstalledRepos } from "../orb/installed-repos-sync";
import { githubRateLimitAdmissionKeyForInstallation, makeInstallationOctokit } from "../github/client";
import { withInstallationTokenRetry } from "../github/app";
import { runScheduledLedgerAnchor, resolveGitAnchorTarget } from "../review/ledger-anchor-scheduler";
import { submitToGitAnchor } from "../review/ledger-anchor-git";
import { incr } from "../selfhost/metrics";
import { generateSignalSnapshots } from "./signal-snapshot";
import { isDecisionAuditEnabled, runDecisionAuditSample } from "../review/decision-audit";
Expand Down Expand Up @@ -93,6 +97,23 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
case "refresh-registry":
await refreshRegistry(env);
return;
case "anchor-decision-ledger": {
// Git anchoring runs only when BOTH the target (owner/repo, #9273) and an installation id (to mint a
// token through) are configured -- unset means that backend simply isn't wired up yet; Rekor still
// proceeds independently either way (runScheduledLedgerAnchor's own posture).
const gitTarget = resolveGitAnchorTarget(env);
const installationId = Number(env.LOOPOVER_LEDGER_ANCHOR_GIT_INSTALLATION_ID);
const submitGit =
gitTarget && Number.isInteger(installationId) && installationId > 0
? async (signed: Parameters<typeof submitToGitAnchor>[1]) =>
withInstallationTokenRetry(env, installationId, async (token) => {
const octokit = makeInstallationOctokit(env, token, "live", githubRateLimitAdmissionKeyForInstallation(installationId));
await submitToGitAnchor(env, signed, octokit, gitTarget);
})
: null;
await runScheduledLedgerAnchor(env, { isHourly: message.isHourly }, { submitGit });
return;
}
case "sync-brokered-installed-repos": {
const syncResult = await syncBrokeredInstalledRepos(env);
// syncBrokeredInstalledRepos is deliberately fail-safe (never throws -- a miss self-heals on the next
Expand Down
24 changes: 14 additions & 10 deletions src/review/decision-record.ts
Original file line number Diff line number Diff line change
Expand Up @@ -335,6 +335,20 @@ export async function ledgerRowHash(prevHash: string, fields: LedgerRowFields):
* by the verify endpoint's record/ledger reconciliation, a follow-up check, rather than by losing the
* decision itself).
*/
/** The chain's current tip -- {@link LEDGER_GENESIS_HASH}/seq 0/count 0 on an empty ledger. Deliberately
* lighter than {@link verifyDecisionLedger} (which additionally walks and verifies a window): a caller that
* only needs "what is the tip right now" (the scheduled anchoring job, #9274) shouldn't pay for a
* self-consistency walk it isn't asking for. */
export async function loadDecisionLedgerTip(env: Env): Promise<{ seq: number; rowHash: string; totalCount: number }> {
const [totalRow, tipRow] = await Promise.all([
env.DB.prepare("SELECT COUNT(*) AS n FROM decision_ledger").first<{ n: number }>(),
env.DB.prepare("SELECT seq, row_hash AS rowHash FROM decision_ledger ORDER BY seq DESC LIMIT 1").first<{ seq: number; rowHash: string }>(),
]);
/* v8 ignore next -- defensive: a bare COUNT(*) always returns exactly one row, mirroring
* verifyDecisionLedger's identical note. */
return { seq: tipRow?.seq ?? 0, rowHash: tipRow?.rowHash ?? LEDGER_GENESIS_HASH, totalCount: totalRow?.n ?? 0 };
}

export async function appendDecisionLedger(env: Env, recordId: string, recordDigest: string, attempts = 3): Promise<void> {
for (let attempt = 1; attempt <= attempts; attempt += 1) {
const tip = await env.DB.prepare("SELECT seq, row_hash AS rowHash FROM decision_ledger ORDER BY seq DESC LIMIT 1").first<{ seq: number; rowHash: string }>();
Expand Down Expand Up @@ -375,16 +389,6 @@ export type LedgerBreak =
// gap/predecessor/hash checks above can see, since those only ever compare ledger rows against each other).
| { kind: "missing_record"; atSeq: number; recordId: string };

/** #9122: the exact shape a scheduled external-anchoring job (git-commit checkpoint, transparency log, or an
* on-chain commitment — the actual publishing mechanism is a genuinely open infra/protocol decision tracked
* on the issue, deliberately NOT built here) would publish for a given tip: enough for a third party to later
* prove "the ledger's tip really was this, at this time" against whatever anchor eventually receives it. Pure
* and synchronous — this module has no scheduler and calls this from nowhere yet; a future cron handler is
* the natural caller, using the tipSeq/tipHash verifyDecisionLedger already returns on every call. */
export function buildLedgerAnchorPayload(tip: { seq: number; rowHash: string }, at: string = nowIso()): { seq: number; rowHash: string; at: string } {
return { seq: tip.seq, rowHash: tip.rowHash, at };
}

/**
* Verify a window of the chain, resumable via `afterSeq` (0 = genesis). Reports the FIRST break with its
* class — a gap, a broken predecessor link, a rewritten row, a short tail, a content mismatch, or a missing
Expand Down
16 changes: 16 additions & 0 deletions src/review/ledger-anchor-persistence.ts
Original file line number Diff line number Diff line change
Expand Up @@ -149,3 +149,19 @@ function parseBackendRef(raw: string | null): unknown {
return null;
}
}

/**
* The most recent anchor ATTEMPT, regardless of backend or status -- the reference point the scheduler
* (#9274) compares the live tip against. Deliberately not "the most recent SUCCESSFUL anchor": if a backend
* is down for a stretch, treating each failed retry as if nothing had been attempted would mean hammering the
* same stale checkpoint every cron tick against a backend that keeps failing, rather than advancing to a
* newer tip as time passes and letting the gap be visible on #9271's own public attempt log. Returns null
* only when nothing has ever been recorded (the very first anchor ever).
*/
export async function loadLastLedgerAnchorAttempt(env: Env): Promise<{ seq: number; rowHash: string } | null> {
const row = await env.DB.prepare("SELECT seq, row_hash AS rowHash FROM decision_ledger_anchors ORDER BY created_at DESC, id DESC LIMIT 1").first<{
seq: number;
rowHash: string;
}>();
return row == null ? null : row;
}
122 changes: 122 additions & 0 deletions src/review/ledger-anchor-scheduler.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
// Scheduled checkpoint anchoring (#9274, epic #9267). Ties #9272 (Rekor) and #9273 (git-commit) together into
// a periodic job -- checkpoint cadence, not per-record, per the mechanism research on #9267: anchoring every
// decision record is wrong on every axis (Rekor is a donated public good; a git commit per PR review is
// noise; a naive per-record job has no natural batching anyway).
import { errorMessage, nowIso } from "../utils/json";
import { buildLedgerAnchorPayload, currentAnchorKey, parseAnchorPublicKeys, signLedgerAnchorPayload, type SignedLedgerAnchor } from "./ledger-anchor";
import { loadDecisionLedgerTip } from "./decision-record";
import { loadLastLedgerAnchorAttempt, recordLedgerAnchorAttempt, type LedgerAnchorBackend } from "./ledger-anchor-persistence";
import { submitToRekor } from "./ledger-anchor-rekor";
import type { LedgerAnchorGitTarget } from "./ledger-anchor-git";

/** Bounds the unanchored window by record count, independent of the hourly clock -- a burst of activity
* cannot sit unanchored for a full hour just because it happened between ticks. */
export const LEDGER_ANCHOR_SEQ_THRESHOLD = 256;

export type LedgerAnchorScheduleDecision =
| { shouldAnchor: false; reason: "empty_ledger" | "unchanged" }
| { shouldAnchor: true; reason: "hourly" | "seq_threshold" };

/**
* Pure scheduling decision: given the live tip and the last attempt (regardless of that attempt's own
* success/failure -- see {@link loadLastLedgerAnchorAttempt}'s own reasoning), should THIS tick anchor?
*
* `seq_threshold` is checked treating a never-anchored ledger as starting from seq 0 -- a burst of >= the
* threshold worth of activity before the very first anchor ever still fires immediately, rather than waiting
* for the next hourly tick just because there is no prior anchor to compare against.
*/
export function decideLedgerAnchorSchedule(input: {
isHourly: boolean;
currentTip: { seq: number; rowHash: string };
lastAnchor: { seq: number; rowHash: string } | null;
seqThreshold: number;
}): LedgerAnchorScheduleDecision {
if (input.currentTip.seq === 0) return { shouldAnchor: false, reason: "empty_ledger" };

const lastAnchorSeq = input.lastAnchor?.seq ?? 0;
if (input.currentTip.seq - lastAnchorSeq >= input.seqThreshold) return { shouldAnchor: true, reason: "seq_threshold" };

const tipUnchanged = input.lastAnchor !== null && input.lastAnchor.rowHash === input.currentTip.rowHash;
if (input.isHourly && !tipUnchanged) return { shouldAnchor: true, reason: "hourly" };

return { shouldAnchor: false, reason: "unchanged" };
}

/** Reads the four `LOOPOVER_LEDGER_ANCHOR_GIT_*` config vars into a target, or `null` when the required
* owner/repo pair is unset -- git anchoring is simply not configured yet, the same honest-degrade posture
* as an unset signing key, not an error. */
export function resolveGitAnchorTarget(env: {
LOOPOVER_LEDGER_ANCHOR_GIT_OWNER?: string;
LOOPOVER_LEDGER_ANCHOR_GIT_REPO?: string;
LOOPOVER_LEDGER_ANCHOR_GIT_BRANCH?: string;
LOOPOVER_LEDGER_ANCHOR_GIT_PATH?: string;
}): LedgerAnchorGitTarget | null {
const owner = env.LOOPOVER_LEDGER_ANCHOR_GIT_OWNER;
const repo = env.LOOPOVER_LEDGER_ANCHOR_GIT_REPO;
if (!owner || !repo) return null;
return { owner, repo, branch: env.LOOPOVER_LEDGER_ANCHOR_GIT_BRANCH || "main", path: env.LOOPOVER_LEDGER_ANCHOR_GIT_PATH || "anchors.jsonl" };
}

export type LedgerAnchorSchedulerDeps = {
/** Defaults to the real Rekor backend. Injectable so a test exercises the scheduler's own decision logic
* without a real network call. */
submitRekor?: (signed: SignedLedgerAnchor, publicKeySpki: string) => Promise<void>;
/** `null`/omitted when git anchoring isn't configured (no target, no installation) -- the scheduler simply
* skips that backend, same posture as an unset signing key. The real caller (the queue processor) wraps
* the actual `submitToGitAnchor` call in `withInstallationTokenRetry` here, so a stale token retries the
* WHOLE attempt with a fresh one, matching every other GitHub write in this engine. */
submitGit?: ((signed: SignedLedgerAnchor) => Promise<void>) | null;
};

/**
* Run one scheduling tick. Never blocks a review: this has no caller in the live decision-record persist
* path at all -- it exists only as a queued background job (#9274's own dispatch wiring) -- and neither
* backend it calls ever rejects (each records its own `status: 'failed'` via #9271 instead of throwing), so
* a total anchoring failure cannot propagate anywhere. Returns the scheduling decision so a caller/test can
* observe WHY nothing happened, without needing a second query.
*/
export async function runScheduledLedgerAnchor(env: Env, options: { isHourly: boolean; now?: string }, deps: LedgerAnchorSchedulerDeps = {}): Promise<LedgerAnchorScheduleDecision> {
const now = options.now ?? nowIso();
const [tip, lastAnchor] = await Promise.all([loadDecisionLedgerTip(env), loadLastLedgerAnchorAttempt(env)]);
const decision = decideLedgerAnchorSchedule({ isHourly: options.isHourly, currentTip: tip, lastAnchor, seqThreshold: LEDGER_ANCHOR_SEQ_THRESHOLD });
if (!decision.shouldAnchor) return decision;

const keys = parseAnchorPublicKeys(env.LOOPOVER_LEDGER_ANCHOR_KEYS);
const current = currentAnchorKey(keys);
if (!current || !env.LOOPOVER_LEDGER_ANCHOR_PRIVATE_KEY) {
console.log(
JSON.stringify({
event: "ledger_anchor_skipped_unconfigured",
reason: !current ? "no_current_signing_key_published" : "no_private_key_configured",
}),
);
return decision;
}

const payload = buildLedgerAnchorPayload(tip, now);
const signed = await signLedgerAnchorPayload(payload, env.LOOPOVER_LEDGER_ANCHOR_PRIVATE_KEY, current.keyId);

const submitRekor = deps.submitRekor ?? ((s, publicKeySpki) => submitToRekor(env, s, publicKeySpki, fetch));
// guardedSubmit is the actual safety boundary, not a convention every injected function is trusted to
// honor: #9272's real submitToRekor and #9273's real submitToGitAnchor never reject on their own, but an
// INJECTED submitGit (the real caller wraps token-minting around submitToGitAnchor, and minting a fresh
// installation token is a genuine IO call that CAN throw, unlike anything inside submitToGitAnchor itself)
// is not something this module controls. Catching here, once, centrally, is what makes "neither backend
// ever rejects" true structurally rather than by every call site's own discipline.
const guardedSubmit = async (backend: LedgerAnchorBackend, submit: () => Promise<void>): Promise<void> => {
try {
await submit();
} catch (error) {
await recordLedgerAnchorAttempt(env, { payload: signed.payload, signature: signed.signature, keyId: signed.keyId, backend, status: "failed", error: errorMessage(error) });
}
};

const attempts: Promise<void>[] = [guardedSubmit("rekor", () => submitRekor(signed, current.publicKeySpki))];
if (deps.submitGit) {
const submitGit = deps.submitGit;
attempts.push(guardedSubmit("git", () => submitGit(signed)));
}
await Promise.all(attempts);

return decision;
}
4 changes: 4 additions & 0 deletions src/selfhost/queue-common.ts
Original file line number Diff line number Diff line change
Expand Up @@ -777,6 +777,10 @@ export function scheduledEnqueueJitterMs(): number {
const IMMEDIATE_SCHEDULED_JOB_TYPES = new Set<string>([
"agent-regate-sweep",
"retry-orb-relay",
// #9274 (epic #9267): checked every ~2-min tick like the sweep above (the seq-threshold anchoring trigger
// must fire independent of the hourly clock), and is a cheap 1-2 query no-op the overwhelming majority of
// ticks -- not a heavy per-repo fan-out parent, so it needs no phase-spreading either.
"anchor-decision-ledger",
]);

export function scheduledEnqueueDelaySeconds(jobType: string): number {
Expand Down
8 changes: 8 additions & 0 deletions src/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,14 @@ export type JobMessage =
type: "refresh-registry";
requestedBy: "schedule" | "api" | "test";
}
| {
// Scheduled decision-ledger anchoring (#9274, epic #9267). `isHourly` is threaded through from the
// dispatcher's own clock computation (src/index.ts) rather than re-derived here, so there is exactly
// ONE place that decides "what time is it" for every scheduled job, this one included.
type: "anchor-decision-ledger";
requestedBy: "schedule" | "test";
isHourly: boolean;
}
| {
type: "sync-brokered-installed-repos";
requestedBy: "schedule" | "api" | "test";
Expand Down
29 changes: 18 additions & 11 deletions test/unit/decision-record.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import {
sha256Hex,
type DecisionRecord,
} from "../../src/review/decision-record";
import { appendDecisionLedger, buildLedgerAnchorPayload, LEDGER_GENESIS_HASH, loadDecisionRecordCollapsible, loadPublicDecisionRecord, verifyDecisionLedger } from "../../src/review/decision-record";
import { appendDecisionLedger, LEDGER_GENESIS_HASH, loadDecisionLedgerTip, loadDecisionRecordCollapsible, loadPublicDecisionRecord, verifyDecisionLedger } from "../../src/review/decision-record";
import { createTestEnv } from "../helpers/d1";

// #8836: the digests are commitments a contributor can challenge — key-order invariance and unicode
Expand Down Expand Up @@ -333,18 +333,25 @@ describe("loadPublicDecisionRecord (#9123)", () => {
});
});

describe("buildLedgerAnchorPayload (#9122)", () => {
it("returns exactly {seq, rowHash, at} from a tip, using a caller-supplied timestamp", () => {
const payload = buildLedgerAnchorPayload({ seq: 5, rowHash: "a".repeat(64) }, "2026-01-01T00:00:00.000Z");
expect(payload).toEqual({ seq: 5, rowHash: "a".repeat(64), at: "2026-01-01T00:00:00.000Z" });
// #9270 superseded the old {seq, rowHash, at} stub that used to live here (buildLedgerAnchorPayload) with a
// self-describing, SIGNED payload -- see src/review/ledger-anchor.ts and test/unit/ledger-anchor.test.ts.

describe("loadDecisionLedgerTip (#9274)", () => {
it("returns genesis/seq 0/count 0 on an empty ledger", async () => {
expect(await loadDecisionLedgerTip(createTestEnv())).toEqual({ seq: 0, rowHash: LEDGER_GENESIS_HASH, totalCount: 0 });
});

it("defaults `at` to the current time when the caller omits it", () => {
const beforeMs = Date.now();
const payload = buildLedgerAnchorPayload({ seq: 1, rowHash: LEDGER_GENESIS_HASH });
expect(payload.seq).toBe(1);
expect(payload.rowHash).toBe(LEDGER_GENESIS_HASH);
expect(Date.parse(payload.at)).toBeGreaterThanOrEqual(beforeMs);
it("returns the real tip and total count once records exist, lighter than a full verify walk", async () => {
const env = createTestEnv();
await appendDecisionLedger(env, "record:acme/widgets#1", "digest1");
await appendDecisionLedger(env, "record:acme/widgets#2", "digest2");
const tip = await loadDecisionLedgerTip(env);
expect(tip.seq).toBe(2);
expect(tip.totalCount).toBe(2);
expect(tip.rowHash).toMatch(/^[0-9a-f]{64}$/);
// Matches what verifyDecisionLedger's own tip computation reports for the same chain.
const verified = await verifyDecisionLedger(env);
expect(tip).toEqual({ seq: verified.tipSeq, rowHash: verified.tipHash, totalCount: verified.totalCount });
});
});

Expand Down
Loading