diff --git a/cmd/entire/cli/checkpoint/migrate.go b/cmd/entire/cli/checkpoint/migrate.go index 1c768cfc8c..96a5127014 100644 --- a/cmd/entire/cli/checkpoint/migrate.go +++ b/cmd/entire/cli/checkpoint/migrate.go @@ -127,7 +127,7 @@ func migrateBranchToRefs( return nil } if alreadyImported { - if err := queue.Enqueue(refName); err != nil { + if err := queue.EnqueueRef(repo, refName); err != nil { return fmt.Errorf("enqueue checkpoint %s for push: %w", cid, err) } result.Skipped++ @@ -176,7 +176,7 @@ func migrateBranchToRefs( // The migration's queued-for-push contract is guaranteed for both new // and already-imported refs. Duplicates collapse on Drain. - if err := queue.Enqueue(refName); err != nil { + if err := queue.EnqueueRef(repo, refName); err != nil { return fmt.Errorf("enqueue checkpoint %s for push: %w", cid, err) } if migrated { diff --git a/cmd/entire/cli/checkpoint/pushqueue.go b/cmd/entire/cli/checkpoint/pushqueue.go index 8fe35dfec8..519dccc411 100644 --- a/cmd/entire/cli/checkpoint/pushqueue.go +++ b/cmd/entire/cli/checkpoint/pushqueue.go @@ -42,7 +42,15 @@ func (q *PushQueue) lock() (*os.Root, func(), error) { // pushQueueEntry is one JSONL record: a checkpoint ref awaiting push. type pushQueueEntry struct { - Ref string `json:"ref"` + Ref string `json:"ref"` + Hash string `json:"hash,omitempty"` +} + +// PushQueueEntry identifies one observed generation of a checkpoint ref. Hash +// is zero only for queue records written by older Entire versions. +type PushQueueEntry struct { + Ref plumbing.ReferenceName + Hash plumbing.Hash } // PushQueue is a flock-protected JSONL list of checkpoint refs awaiting push, @@ -78,13 +86,24 @@ func PushQueueForRepo(_ context.Context, repo *git.Repository) (*PushQueue, erro // idempotent. Enqueue takes the lock so concurrent writers never interleave a // partial line. func (q *PushQueue) Enqueue(ref plumbing.ReferenceName) error { + return q.EnqueueEntry(PushQueueEntry{Ref: ref}) +} + +// EnqueueEntry appends an observed ref generation. A later generation of the +// same ref supersedes it when the queue is read, while exact-generation removal +// cannot erase the later entry. +func (q *PushQueue) EnqueueEntry(entry PushQueueEntry) error { root, release, err := q.lock() if err != nil { return err } defer release() - line, err := json.Marshal(pushQueueEntry{Ref: ref.String()}) + diskEntry := pushQueueEntry{Ref: entry.Ref.String()} + if !entry.Hash.IsZero() { + diskEntry.Hash = entry.Hash.String() + } + line, err := json.Marshal(diskEntry) if err != nil { return fmt.Errorf("encode push queue entry: %w", err) } @@ -99,6 +118,18 @@ func (q *PushQueue) Enqueue(ref plumbing.ReferenceName) error { return nil } +// EnqueueRef records the ref's current observed generation. When the ref +// cannot be resolved it still queues the ref, with no generation: a ref that +// misses the queue is never pushed, while a generation-less entry is the legacy +// form delivery already re-resolves (and prunes if the ref is gone). +func (q *PushQueue) EnqueueRef(repo *git.Repository, refName plumbing.ReferenceName) error { + entry := PushQueueEntry{Ref: refName} + if ref, err := repo.Reference(refName, true); err == nil { + entry.Hash = ref.Hash() + } + return q.EnqueueEntry(entry) +} + // Drain returns the de-duplicated refs currently queued, in first-seen order. It // does NOT remove them; call Remove after a confirmed push so a failed push // retries next time. A missing queue file yields no refs. @@ -110,22 +141,29 @@ func (q *PushQueue) Enqueue(ref plumbing.ReferenceName) error { // compaction point (e.g. a long-lived session that keeps re-enqueuing the same // checkpoint ref but never pushes). func (q *PushQueue) Drain() ([]plumbing.ReferenceName, error) { + entries, err := q.DrainEntries() + return pushQueueRefs(entries), err +} + +// DrainEntries returns the latest observed generation of every queued ref in +// first-seen ref order. It compacts redundant, superseded, and malformed lines. +func (q *PushQueue) DrainEntries() ([]PushQueueEntry, error) { root, release, err := q.lock() if err != nil { return nil, err } defer release() - refs, rawLines, err := q.readLocked(root) + entries, rawLines, err := q.readEntriesLocked(root) if err != nil { return nil, err } - if rawLines > len(refs) { - if err := q.rewriteLocked(root, refs); err != nil { + if rawLines > len(entries) { + if err := q.rewriteEntriesLocked(root, entries); err != nil { return nil, err } } - return refs, nil + return entries, nil } // Peek returns the de-duplicated refs currently queued, in first-seen order, @@ -133,14 +171,20 @@ func (q *PushQueue) Drain() ([]plumbing.ReferenceName, error) { // compacts redundant lines in place) — for counters/status displays that must // observe the queue without owning a push. A missing queue file yields no refs. func (q *PushQueue) Peek() ([]plumbing.ReferenceName, error) { + entries, err := q.PeekEntries() + return pushQueueRefs(entries), err +} + +// PeekEntries is the read-only counterpart to DrainEntries. +func (q *PushQueue) PeekEntries() ([]PushQueueEntry, error) { root, release, err := q.lock() if err != nil { return nil, err } defer release() - refs, _, err := q.readLocked(root) - return refs, err + entries, _, err := q.readEntriesLocked(root) + return entries, err } // Remove deletes the given refs from the queue, preserving any entries appended @@ -156,7 +200,7 @@ func (q *PushQueue) Remove(refs []plumbing.ReferenceName) error { } defer release() - current, _, err := q.readLocked(root) + current, _, err := q.readEntriesLocked(root) if err != nil { return err } @@ -164,20 +208,51 @@ func (q *PushQueue) Remove(refs []plumbing.ReferenceName) error { for _, r := range refs { removed[r.String()] = struct{}{} } - kept := make([]plumbing.ReferenceName, 0, len(current)) - for _, r := range current { - if _, drop := removed[r.String()]; drop { + kept := make([]PushQueueEntry, 0, len(current)) + for _, entry := range current { + if _, drop := removed[entry.Ref.String()]; drop { continue } - kept = append(kept, r) + kept = append(kept, entry) } - return q.rewriteLocked(root, kept) + return q.rewriteEntriesLocked(root, kept) } -// Rotate moves the given refs to the back of the queue, keeping the relative -// order within both the moved and the untouched group. Refs not currently queued -// are ignored, and entries appended after a Drain keep their place ahead of the -// rotated ones. +// RemoveEntries removes only exact ref generations. A newer generation of the +// same ref therefore survives cleanup for a push that delivered an older one. +func (q *PushQueue) RemoveEntries(entries []PushQueueEntry) error { + if len(entries) == 0 { + return nil + } + root, release, err := q.lock() + if err != nil { + return err + } + defer release() + + current, _, err := q.readEntriesLocked(root) + if err != nil { + return err + } + removed := make(map[PushQueueEntry]struct{}, len(entries)) + for _, entry := range entries { + removed[entry] = struct{}{} + } + kept := make([]PushQueueEntry, 0, len(current)) + for _, entry := range current { + if _, drop := removed[entry]; drop { + continue + } + kept = append(kept, entry) + } + return q.rewriteEntriesLocked(root, kept) +} + +// Rotate moves the given queue entries to the back of the queue, keeping the +// relative order within both the moved and the untouched group. Entries match +// by exact generation (ref and hash), so a newer generation of the same ref +// enqueued meanwhile keeps its place. Entries not currently queued are ignored, +// and entries appended after a Drain keep their place ahead of the rotated ones. // // This is what keeps a bounded flush fair. The queue is drained in first-seen // order and a flush that stops early leaves the rest for the next push, so @@ -189,8 +264,8 @@ func (q *PushQueue) Remove(refs []plumbing.ReferenceName) error { // One lock for the whole reorder, rather than Remove followed by Enqueue: that // pair leaves a window where the refs are in neither the queue nor the remote, // and a crash inside it drops them. -func (q *PushQueue) Rotate(refs []plumbing.ReferenceName) error { - if len(refs) == 0 { +func (q *PushQueue) Rotate(entries []PushQueueEntry) error { + if len(entries) == 0 { return nil } root, release, err := q.lock() @@ -199,43 +274,47 @@ func (q *PushQueue) Rotate(refs []plumbing.ReferenceName) error { } defer release() - current, _, err := q.readLocked(root) + current, _, err := q.readEntriesLocked(root) if err != nil { return err } - rotating := make(map[string]struct{}, len(refs)) - for _, r := range refs { - rotating[r.String()] = struct{}{} + rotating := make(map[PushQueueEntry]struct{}, len(entries)) + for _, entry := range entries { + rotating[entry] = struct{}{} } - kept := make([]plumbing.ReferenceName, 0, len(current)) - moved := make([]plumbing.ReferenceName, 0, len(refs)) - for _, r := range current { - if _, rotate := rotating[r.String()]; rotate { - moved = append(moved, r) + kept := make([]PushQueueEntry, 0, len(current)) + moved := make([]PushQueueEntry, 0, len(entries)) + for _, entry := range current { + if _, rotate := rotating[entry]; rotate { + moved = append(moved, entry) continue } - kept = append(kept, r) + kept = append(kept, entry) } if len(moved) == 0 { return nil } - return q.rewriteLocked(root, append(kept, moved...)) + return q.rewriteEntriesLocked(root, append(kept, moved...)) } -// rewriteLocked replaces the queue file with exactly refs (de-duplicated, one +// rewriteEntriesLocked replaces the queue file with exactly entries (de-duplicated, one // line each), or removes the file when refs is empty so a clean repo has no // stray queue. The caller must hold the lock. The write is atomic (temp file + // rename) so a concurrent reader never sees a half-written queue. -func (q *PushQueue) rewriteLocked(root *os.Root, refs []plumbing.ReferenceName) error { - if len(refs) == 0 { +func (q *PushQueue) rewriteEntriesLocked(root *os.Root, entries []PushQueueEntry) error { + if len(entries) == 0 { if err := root.Remove(pushQueueFileName); err != nil && !os.IsNotExist(err) { return fmt.Errorf("remove empty push queue: %w", err) } return nil } var buf bytes.Buffer - for _, r := range refs { - line, err := json.Marshal(pushQueueEntry{Ref: r.String()}) + for _, entry := range entries { + diskEntry := pushQueueEntry{Ref: entry.Ref.String()} + if !entry.Hash.IsZero() { + diskEntry.Hash = entry.Hash.String() + } + line, err := json.Marshal(diskEntry) if err != nil { return fmt.Errorf("encode push queue entry: %w", err) } @@ -248,7 +327,7 @@ func (q *PushQueue) rewriteLocked(root *os.Root, refs []plumbing.ReferenceName) return nil } -// readLocked parses the queue file into de-duplicated refs, preserving first-seen +// readEntriesLocked parses the queue file into de-duplicated entries, preserving first-seen // order. The caller must hold the lock. Malformed lines are skipped rather than // failing the whole drain — a single bad record must not strand every queued ref. // @@ -256,7 +335,7 @@ func (q *PushQueue) rewriteLocked(root *os.Root, refs []plumbing.ReferenceName) // malformed records), so callers can detect when the file holds more than the // de-duplicated set and is worth compacting: rawLines > len(refs) exactly when // there were redundant lines. -func (q *PushQueue) readLocked(root *os.Root) (refs []plumbing.ReferenceName, rawLines int, err error) { +func (q *PushQueue) readEntriesLocked(root *os.Root) (entries []PushQueueEntry, rawLines int, err error) { f, err := root.Open(pushQueueFileName) if err != nil { if os.IsNotExist(err) { @@ -266,7 +345,7 @@ func (q *PushQueue) readLocked(root *os.Root) (refs []plumbing.ReferenceName, ra } defer f.Close() - seen := make(map[string]struct{}) + seen := make(map[string]int) scanner := bufio.NewScanner(f) scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) for scanner.Scan() { @@ -279,16 +358,31 @@ func (q *PushQueue) readLocked(root *os.Root) (refs []plumbing.ReferenceName, ra if err := json.Unmarshal(line, &entry); err != nil || entry.Ref == "" { continue } - if _, dup := seen[entry.Ref]; dup { + parsed := PushQueueEntry{Ref: plumbing.ReferenceName(entry.Ref)} + // A corrupt generation keeps the ref, generation-less: dropping the line + // would compact it away, and the queue is the only push discovery. + if plumbing.IsHash(entry.Hash) { + parsed.Hash = plumbing.NewHash(entry.Hash) + } + if idx, dup := seen[entry.Ref]; dup { + entries[idx] = parsed continue } - seen[entry.Ref] = struct{}{} - refs = append(refs, plumbing.ReferenceName(entry.Ref)) + seen[entry.Ref] = len(entries) + entries = append(entries, parsed) } if err := scanner.Err(); err != nil { return nil, 0, fmt.Errorf("read push queue: %w", err) } - return refs, rawLines, nil + return entries, rawLines, nil +} + +func pushQueueRefs(entries []PushQueueEntry) []plumbing.ReferenceName { + refs := make([]plumbing.ReferenceName, 0, len(entries)) + for _, entry := range entries { + refs = append(refs, entry.Ref) + } + return refs } // writeQueueAtomic writes data to a temp file inside root and renames it over diff --git a/cmd/entire/cli/checkpoint/pushqueue_test.go b/cmd/entire/cli/checkpoint/pushqueue_test.go index 9210854417..794f9673e5 100644 --- a/cmd/entire/cli/checkpoint/pushqueue_test.go +++ b/cmd/entire/cli/checkpoint/pushqueue_test.go @@ -1,11 +1,14 @@ package checkpoint import ( + "encoding/json" "os" "path/filepath" "strings" "testing" + "github.com/entireio/cli/cmd/entire/cli/gitrepo" + "github.com/entireio/cli/cmd/entire/cli/testutil" "github.com/go-git/go-git/v6/plumbing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -208,20 +211,22 @@ func TestPushQueue_RotateMovesRefsToBack(t *testing.T) { require.NoError(t, q.Enqueue(ref)) } - require.NoError(t, q.Rotate([]plumbing.ReferenceName{a, b})) + entry := func(ref plumbing.ReferenceName) PushQueueEntry { return PushQueueEntry{Ref: ref} } + + require.NoError(t, q.Rotate([]PushQueueEntry{entry(a), entry(b)})) refs, err := q.Drain() require.NoError(t, err) assert.Equal(t, []plumbing.ReferenceName{c, a, b}, refs, "rotated refs go to the back, both groups keeping their relative order") // A ref that is not queued is ignored rather than added. - require.NoError(t, q.Rotate([]plumbing.ReferenceName{mustRefName(t, "ffffffffffff")})) + require.NoError(t, q.Rotate([]PushQueueEntry{entry(mustRefName(t, "ffffffffffff"))})) refs, err = q.Drain() require.NoError(t, err) assert.Equal(t, []plumbing.ReferenceName{c, a, b}, refs, "an absent ref must not join the queue") // Rotating everything is a no-op on order, and never drops a ref. - require.NoError(t, q.Rotate([]plumbing.ReferenceName{c, a, b})) + require.NoError(t, q.Rotate([]PushQueueEntry{entry(c), entry(a), entry(b)})) refs, err = q.Drain() require.NoError(t, err) assert.Equal(t, []plumbing.ReferenceName{c, a, b}, refs) @@ -232,4 +237,120 @@ func TestPushQueue_RotateMovesRefsToBack(t *testing.T) { assert.Len(t, refs, 3, "rotation only reorders; it never removes") } +func TestPushQueue_RemovePreservesNewerGenerationOfSameRef(t *testing.T) { + t.Parallel() + q := NewPushQueue(t.TempDir()) + ref := mustRefName(t, "a1b2c3d4e5f6") + old := PushQueueEntry{Ref: ref, Hash: plumbing.NewHash(strings.Repeat("a", 40))} + newer := PushQueueEntry{Ref: ref, Hash: plumbing.NewHash(strings.Repeat("b", 40))} + + require.NoError(t, q.EnqueueEntry(old)) + require.NoError(t, q.EnqueueEntry(newer)) + require.NoError(t, q.RemoveEntries([]PushQueueEntry{old})) + + entries, err := q.DrainEntries() + require.NoError(t, err) + assert.Equal(t, []PushQueueEntry{newer}, entries, + "cleanup for an older generation must not remove the newer generation") +} + +func TestPushQueue_DrainKeepsLatestGenerationAtFirstSeenPosition(t *testing.T) { + t.Parallel() + q := NewPushQueue(t.TempDir()) + refA := mustRefName(t, "a1b2c3d4e5f6") + refB := mustRefName(t, "b2c3d4e5f6a1") + a1 := PushQueueEntry{Ref: refA, Hash: plumbing.NewHash(strings.Repeat("a", 40))} + b1 := PushQueueEntry{Ref: refB, Hash: plumbing.NewHash(strings.Repeat("b", 40))} + a2 := PushQueueEntry{Ref: refA, Hash: plumbing.NewHash(strings.Repeat("c", 40))} + + require.NoError(t, q.EnqueueEntry(a1)) + require.NoError(t, q.EnqueueEntry(b1)) + require.NoError(t, q.EnqueueEntry(a2)) + + entries, err := q.DrainEntries() + require.NoError(t, err) + assert.Equal(t, []PushQueueEntry{a2, b1}, entries, + "latest generation must win without moving the ref behind later refs") +} + +func TestPushQueue_LegacyEntryCannotRemoveKnownGeneration(t *testing.T) { + t.Parallel() + q := NewPushQueue(t.TempDir()) + ref := mustRefName(t, "a1b2c3d4e5f6") + legacy := PushQueueEntry{Ref: ref} + known := PushQueueEntry{Ref: ref, Hash: plumbing.NewHash(strings.Repeat("d", 40))} + + root, release, err := q.lock() + require.NoError(t, err) + legacyLine, err := json.Marshal(pushQueueEntry{Ref: ref.String()}) + require.NoError(t, err) + require.NoError(t, writeQueueAtomic(root, append(legacyLine, '\n'))) + release() + + require.NoError(t, q.EnqueueEntry(known)) + require.NoError(t, q.RemoveEntries([]PushQueueEntry{legacy})) + + entries, err := q.DrainEntries() + require.NoError(t, err) + assert.Equal(t, []PushQueueEntry{known}, entries, + "exact legacy cleanup must not remove a later hash-bearing generation") +} + +// Rotation matches exact generations: a failed older generation must not drag +// a newer generation of the same ref, enqueued meanwhile, to the back. +func TestPushQueue_RotateLeavesNewerGenerationInPlace(t *testing.T) { + t.Parallel() + q := NewPushQueue(t.TempDir()) + refA := mustRefName(t, "a1b2c3d4e5f6") + refB := mustRefName(t, "b2c3d4e5f6a1") + a1 := PushQueueEntry{Ref: refA, Hash: plumbing.NewHash(strings.Repeat("a", 40))} + b1 := PushQueueEntry{Ref: refB, Hash: plumbing.NewHash(strings.Repeat("b", 40))} + a2 := PushQueueEntry{Ref: refA, Hash: plumbing.NewHash(strings.Repeat("c", 40))} + require.NoError(t, q.EnqueueEntry(a1)) + require.NoError(t, q.EnqueueEntry(b1)) + require.NoError(t, q.EnqueueEntry(a2)) + + require.NoError(t, q.Rotate([]PushQueueEntry{a1})) + + entries, err := q.DrainEntries() + require.NoError(t, err) + assert.Equal(t, []PushQueueEntry{a2, b1}, entries, + "the newer generation keeps the ref's first-seen place") +} + +// A ref that misses the queue is never pushed, so a ref EnqueueRef cannot +// resolve is still queued, in the generation-less form delivery re-resolves. +func TestPushQueue_EnqueueRefQueuesAnUnresolvableRef(t *testing.T) { + t.Parallel() + dir := t.TempDir() + testutil.InitRepo(t, dir) + repo, err := gitrepo.OpenPath(dir) + require.NoError(t, err) + q := NewPushQueue(t.TempDir()) + missing := mustRefName(t, "a1b2c3d4e5f6") + + require.NoError(t, q.EnqueueRef(repo, missing)) + + entries, err := q.PeekEntries() + require.NoError(t, err) + assert.Equal(t, []PushQueueEntry{{Ref: missing}}, entries) +} + +// A line whose ref is well formed but whose generation is corrupt keeps the +// ref: compaction would otherwise erase it, and nothing else rediscovers it. +func TestPushQueue_CorruptHashKeepsTheRef(t *testing.T) { + t.Parallel() + q := NewPushQueue(t.TempDir()) + a := mustRefName(t, "a1b2c3d4e5f6") + require.NoError(t, os.WriteFile(q.queuePath(), + []byte(`{"ref":"`+a.String()+`","hash":"not-a-hash"}`+"\n"), 0o600)) + + entries, err := q.DrainEntries() + require.NoError(t, err) + assert.Equal(t, []PushQueueEntry{{Ref: a}}, entries) + entries, err = q.PeekEntries() + require.NoError(t, err) + assert.Equal(t, []PushQueueEntry{{Ref: a}}, entries, "compaction must not erase the ref") +} + func (q *PushQueue) queuePath() string { return filepath.Join(q.dir, pushQueueFileName) } diff --git a/cmd/entire/cli/checkpoint/refs_store.go b/cmd/entire/cli/checkpoint/refs_store.go index 90c4893069..dd8691047d 100644 --- a/cmd/entire/cli/checkpoint/refs_store.go +++ b/cmd/entire/cli/checkpoint/refs_store.go @@ -213,7 +213,7 @@ func (s *gitRefsStore) enqueueForPush(ctx context.Context, refName plumbing.Refe slog.String("ref", refName.String()), slog.String("error", err.Error())) return } - if err := q.Enqueue(refName); err != nil { + if err := q.EnqueueRef(s.repo, refName); err != nil { logging.Warn(ctx, "checkpoint: enqueue checkpoint ref for push failed", slog.String("ref", refName.String()), slog.String("error", err.Error())) } diff --git a/cmd/entire/cli/doctor_migrate.go b/cmd/entire/cli/doctor_migrate.go index 74c2da4f27..9d3291979e 100644 --- a/cmd/entire/cli/doctor_migrate.go +++ b/cmd/entire/cli/doctor_migrate.go @@ -126,15 +126,22 @@ func pushMigratedRefs(ctx context.Context, out io.Writer, repo *git.Repository, pushed, pushDisabled, err := strategy.PushQueuedCheckpointRefs(ctx, repo, pushRemote) if err != nil { + // Partial success is now reachable: refs that already carried the OPF + // trailer ship even when a sibling is still being redacted or the push + // itself failed part-way. Report what landed before the error, or the + // user is told their push failed and never learns what did arrive. + if pushed > 0 { + fmt.Fprintf(out, "Pushed %d checkpoint ref(s).\n", pushed) + } if errors.Is(err, context.Canceled) { return NewSilentError(err) } // Ctrl-C at the OPF prompt is the same gesture as declining the push - // prompt, and lands in the same place: nothing shipped, refs still + // prompt, and lands in the same place: nothing shipped, the refs stay // queued. confirmDoctorFix reports that as a clean decline, so this // must not report it as a failure. if errors.Is(err, strategy.ErrOPFAbortedByUser) { - fmt.Fprintln(out, "OPF cancelled; refs stay queued for the next push.") + fmt.Fprintln(out, "OPF cancelled; the refs stay queued for the next push.") return nil } return fmt.Errorf("push migrated refs: %w", err) diff --git a/cmd/entire/cli/strategy/manual_commit_opf_cas_test.go b/cmd/entire/cli/strategy/manual_commit_opf_cas_test.go index 77b8cf097d..2d47c51379 100644 --- a/cmd/entire/cli/strategy/manual_commit_opf_cas_test.go +++ b/cmd/entire/cli/strategy/manual_commit_opf_cas_test.go @@ -12,9 +12,11 @@ import ( "testing" "time" + "github.com/entireio/cli/cmd/entire/cli/checkpoint" "github.com/entireio/cli/cmd/entire/cli/gitrepo" "github.com/entireio/cli/cmd/entire/cli/paths" "github.com/entireio/cli/cmd/entire/cli/testutil" + "github.com/go-git/go-git/v6" "github.com/go-git/go-git/v6/plumbing" "github.com/stretchr/testify/require" ) @@ -103,6 +105,42 @@ func TestOPFCheckpointRefRewriteDoesNotLoseConcurrentCheckpointUpdate(t *testing "the OPF rewrite must report ref contention while a native update owns the ref lock") } +func TestUpdateOPFRewrittenRefQueuesTheGenerationInstalledByCAS(t *testing.T) { + _, repo, refs := setupGitRefsOPFRepo(t, "a1b2c3d4e5f6") + refName := refs[0] + oldRef, err := repo.Reference(refName, true) + require.NoError(t, err) + oldCommit, err := repo.CommitObject(oldRef.Hash()) + require.NoError(t, err) + rewrittenTip := makeOrphanCommit(t, repo, oldCommit.TreeHash, []plumbing.Hash{oldRef.Hash()}, "OPF rewrite") + concurrentTip := makeOrphanCommit(t, repo, oldCommit.TreeHash, []plumbing.Hash{rewrittenTip}, "concurrent checkpoint write") + queue := &advancingOPFQueue{repo: repo, refName: refName, concurrentTip: concurrentTip} + + err = updateOPFRewrittenRef(t.Context(), repo, queue, refName, rewrittenTip, oldRef.Hash()) + + require.NoError(t, err) + finalRef, err := repo.Reference(refName, true) + require.NoError(t, err) + require.Equal(t, concurrentTip, finalRef.Hash(), "the concurrent checkpoint generation remains current") + require.Equal(t, checkpoint.PushQueueEntry{Ref: refName, Hash: rewrittenTip}, queue.entry, + "the queue records the generation installed by CAS, not the later ref tip") +} + +type advancingOPFQueue struct { + repo *git.Repository + refName plumbing.ReferenceName + concurrentTip plumbing.Hash + entry checkpoint.PushQueueEntry +} + +func (q *advancingOPFQueue) EnqueueEntry(entry checkpoint.PushQueueEntry) error { + if err := q.repo.Storer.SetReference(plumbing.NewHashReference(q.refName, q.concurrentTip)); err != nil { + return err + } + q.entry = entry + return nil +} + func prepareNativeRefUpdate( t *testing.T, repoRoot string, diff --git a/cmd/entire/cli/strategy/manual_commit_opf_refs.go b/cmd/entire/cli/strategy/manual_commit_opf_refs.go index af5205f3e0..7cc7fbc715 100644 --- a/cmd/entire/cli/strategy/manual_commit_opf_refs.go +++ b/cmd/entire/cli/strategy/manual_commit_opf_refs.go @@ -1,15 +1,21 @@ // Pre-push OPF rewrite for the git-refs checkpoint backend, the sibling of -// manual_commit_opf_rewrite.go's entire/checkpoints/v1 rewrite. Both run the -// OPF-augmented redaction once per push; only discovery and ref update differ. +// manual_commit_opf_rewrite.go's entire/checkpoints/v1 rewrite. Beyond +// discovery and ref update, the backends differ in scope: this one redacts one +// queued ref per OPF call, because separate refs are independent chains, while +// the v1 rewrite must take its whole unpushed chain as a unit (each rebuilt +// commit is the next one's parent). That difference is documented at the v1 +// rewrite's own cap check. package strategy import ( "context" "errors" "fmt" + "log/slog" "slices" "github.com/entireio/cli/cmd/entire/cli/checkpoint" + "github.com/entireio/cli/cmd/entire/cli/logging" "github.com/entireio/cli/cmd/entire/cli/trailers" "github.com/entireio/cli/redact" "github.com/go-git/go-git/v6" @@ -29,22 +35,38 @@ import ( // already carrying the trailer (unappliedAncestry) — bounded by the same // resolveBootstrapLimit the v1 path uses. // +// Each ref is collected, cap-checked, scanned, rebuilt and CAS-updated before +// the next ref is loaded. Both the raw-memory ceiling and the prose-leaf +// inference cap therefore apply per ref. A ref that exceeds either is left +// untouched and still queued rather than failing the refs beside it. +// +// That isolation is not unconditional. Two conditions stop the whole flush and +// leave every ref after them untouched: an empty effective category set, and a +// tripped OPF circuit breaker — both mean OPF cannot scan anything, so no +// remaining ref may be stamped applied. See the breaker check in the loop. +// // Caller checks redact.OPFEnabled() and skips this when OPF is off. Returns -// the same error taxonomy as RewriteUnpushedV1WithOPF; the caller fails closed -// by withholding the flush (see prePushCheckpointRefs). +// the same error taxonomy as RewriteUnpushedV1WithOPF; delivery fails closed by +// withholding any ref whose current generation still lacks the trailer. func RewriteQueuedCheckpointRefsWithOPF(ctx context.Context, repo *git.Repository) error { queue, err := checkpoint.PushQueueForRepo(ctx, repo) if err != nil { return fmt.Errorf("resolve push queue: %w", err) } // Peek, not Drain: flushCheckpointRefsQueue owns draining and pruning. - queued, err := queue.Peek() + queuedEntries, err := queue.PeekEntries() if err != nil { return fmt.Errorf("peek push queue: %w", err) } - if len(queued) == 0 { + if len(queuedEntries) == 0 { return nil } + queued := make([]plumbing.ReferenceName, 0, len(queuedEntries)) + entryByRef := make(map[plumbing.ReferenceName]checkpoint.PushQueueEntry, len(queuedEntries)) + for _, entry := range queuedEntries { + queued = append(queued, entry.Ref) + entryByRef[entry.Ref] = entry + } // Both up-front fail-closed gates, in the same order as the v1 rewrite, so // a misconfigured category set surfaces config remediation rather than @@ -56,119 +78,188 @@ func RewriteQueuedCheckpointRefsWithOPF(ctx context.Context, repo *git.Repositor return &OPFRuntimeFailedError{OPFCommand: redact.OPFCommand()} } - // Pass 1: collect every redactable blob from every un-applied commit on every - // queued ref, bounding raw bytes in memory exactly as the v1 collect pass - // does — cumulatively across the whole flush, not per ref. - type pendingCommit struct { - commit *object.Commit - // blobs and paths are parallel; startIdx is this commit's offset into - // the global redacted slice. - blobs []redact.NamedBlob - paths []string - startIdx int - } - type pendingRef struct { - ref plumbing.ReferenceName - old plumbing.Hash - // base is the parent the deepest rewritten commit keeps. - base plumbing.Hash - commits []pendingCommit // ancestor-first - } - var globalBlobs []redact.NamedBlob - pendings := make([]pendingRef, 0, len(queued)) - rawCap := scaleBatchLimit(resolveBatchLimit(), rawByteCapMultiplier) + batchLimit := resolveBatchLimit() + rawCap := rawByteCapForBatchLimit(batchLimit) bootstrapLimit := resolveBootstrapLimit() - var rawBytesSoFar int // Stale entries (refs no longer present locally) are skipped, not pruned: // the queue belongs to the flush. existing, _ := partitionLocalRefs(repo, queued) + var firstErr error for _, refName := range existing { - ref, refErr := repo.Reference(refName, true) - if refErr != nil { - if errors.Is(refErr, plumbing.ErrReferenceNotFound) { + // Whole-flush stop, alongside ErrOPFNoEnabledCategories below: a + // tripped process-wide breaker (this loop's own prior ref, or anything + // earlier in the process) makes BatchBytesWithPrivacyFilter return + // regex-only content with a NIL error, so break — not continue. Skipping + // only this ref would rebuild every later one from content OPF never saw + // and stamp Entire-OPF-Applied on it. Per-ref isolation covers per-ref + // conditions (the size cap); a broken runtime is broken for the whole + // process. Refs already rewritten above were scanned before the trip. + if redact.OPFBreakerTripped() { + // Replaces any earlier per-ref error: a broken runtime is what + // withholds every remaining ref, so it is the error to report. + firstErr = &OPFRuntimeFailedError{OPFCommand: redact.OPFCommand()} + break + } + + pending, collectErr := collectCheckpointRefForOPF(repo, refName, rawCap, bootstrapLimit) + if collectErr != nil { + if errors.Is(collectErr, errNoCheckpointRefOPFWork) { continue } - return fmt.Errorf("resolve checkpoint ref %s: %w", refName, refErr) - } - chain, base, walkErr := unappliedAncestry(repo, ref.Hash()) - if walkErr != nil { - return fmt.Errorf("walk ancestry of %s: %w", refName, walkErr) + if firstErr == nil { + firstErr = collectErr + } + continue } - if len(chain) == 0 { + rewriteErr := rewriteCollectedCheckpointRefWithOPF(ctx, repo, queue, pending, batchLimit) + if rewriteErr == nil { continue } - if len(chain) > bootstrapLimit { - return &BootstrapTooLargeError{Count: len(chain), Limit: bootstrapLimit} + if errors.Is(rewriteErr, redact.ErrOPFNoEnabledCategories) { + return &OPFNoCategoriesError{} } - pr := pendingRef{ref: refName, old: ref.Hash(), base: base} - for _, c := range chain { - tree, treeErr := repo.TreeObject(c.TreeHash) - if treeErr != nil { - return fmt.Errorf("load tree for %s: %w", c.Hash.String()[:7], treeErr) - } - pc := pendingCommit{commit: c, startIdx: len(globalBlobs)} - if err := collectTreeBlobs(repo, tree, "", &pc.blobs, &pc.paths); err != nil { - return fmt.Errorf("collect blobs %s: %w", c.Hash.String()[:7], err) - } - for _, b := range pc.blobs { - rawBytesSoFar += len(b.Content) - } - if rawBytesSoFar > rawCap { - return &OPFRawBytesTooLargeError{RawBytes: rawBytesSoFar, Limit: rawCap} + var runtimeErr *OPFRuntimeFailedError + if errors.As(rewriteErr, &runtimeErr) { + firstErr = rewriteErr // takes precedence, as for the breaker above + // Queue order is stable across pushes, so a ref whose content + // breaks the runtime would otherwise be hit first every time and + // withhold every ref behind it. Moving it to the back lets the next + // push reach those refs first. Best-effort: a lost rotation only + // keeps today's order. + if rotateErr := queue.Rotate([]checkpoint.PushQueueEntry{entryByRef[refName]}); rotateErr != nil { + logging.Warn(ctx, "opf: could not move failed checkpoint ref to the back of the push queue", + slog.String("ref", refName.String()), slog.String("error", rotateErr.Error())) } - globalBlobs = append(globalBlobs, pc.blobs...) - pr.commits = append(pr.commits, pc) + break + } + if firstErr == nil { + firstErr = rewriteErr } - pendings = append(pendings, pr) } - if len(pendings) == 0 { - return nil + return firstErr +} + +type pendingOPFCommit struct { + commit *object.Commit + // blobs and paths are parallel and index the same redaction results. + blobs []redact.NamedBlob + paths []string +} + +type pendingOPFRef struct { + ref plumbing.ReferenceName + old plumbing.Hash + // base is the parent the deepest rewritten commit keeps. + base plumbing.Hash + commits []pendingOPFCommit // ancestor-first +} + +var errNoCheckpointRefOPFWork = errors.New("checkpoint ref does not need OPF rewrite") + +func collectCheckpointRefForOPF( + repo *git.Repository, + refName plumbing.ReferenceName, + rawCap, bootstrapLimit int, +) (*pendingOPFRef, error) { + ref, err := repo.Reference(refName, true) + if err != nil { + if errors.Is(err, plumbing.ErrReferenceNotFound) { + return nil, errNoCheckpointRefOPFWork + } + return nil, fmt.Errorf("resolve checkpoint ref %s: %w", refName, err) + } + chain, base, err := unappliedAncestry(repo, ref.Hash()) + if err != nil { + return nil, fmt.Errorf("walk ancestry of %s: %w", refName, err) + } + if len(chain) == 0 { + return nil, errNoCheckpointRefOPFWork + } + if len(chain) > bootstrapLimit { + return nil, &BootstrapTooLargeError{Count: len(chain), Limit: bootstrapLimit} } - // Pass 2: enforce the leaf-byte cap, then make exactly ONE OPF shell-out - // for the whole flush. - var globalRedacted [][]byte - if len(globalBlobs) > 0 { - leafBytes := redact.SumProseLeafBytes(globalBlobs) - if limit := resolveBatchLimit(); leafBytes > limit { - return &OPFBatchTooLargeError{LeafBytes: leafBytes, Limit: limit} + pending := &pendingOPFRef{ref: refName, old: ref.Hash(), base: base} + rawBudget := newOPFRawByteBudget(rawCap) + for _, commit := range chain { + tree, treeErr := repo.TreeObject(commit.TreeHash) + if treeErr != nil { + return nil, fmt.Errorf("load tree for %s: %w", commit.Hash.String()[:7], treeErr) + } + pc := pendingOPFCommit{commit: commit} + if err := collectTreeBlobsWithinBudget(repo, tree, "", &pc.blobs, &pc.paths, rawBudget); err != nil { + return nil, fmt.Errorf("collect blobs %s: %w", commit.Hash.String()[:7], err) } - globalRedacted, err = redact.BatchBytesWithPrivacyFilter(ctx, globalBlobs) + pending.commits = append(pending.commits, pc) + } + return pending, nil +} + +func rewriteCollectedCheckpointRefWithOPF( + ctx context.Context, + repo *git.Repository, + queue *checkpoint.PushQueue, + pending *pendingOPFRef, + batchLimit int, +) error { + var blobs []redact.NamedBlob + for _, commit := range pending.commits { + blobs = append(blobs, commit.blobs...) + } + if leafBytes := redact.SumProseLeafBytes(blobs); leafBytes > batchLimit { + return &OPFBatchTooLargeError{LeafBytes: leafBytes, Limit: batchLimit} + } + + var redacted [][]byte + if len(blobs) > 0 { + var err error + redacted, err = redact.BatchBytesWithPrivacyFilter(ctx, blobs) if err != nil { if errors.Is(err, redact.ErrOPFNoEnabledCategories) { - return &OPFNoCategoriesError{} + return fmt.Errorf("scan checkpoint ref %s: %w", pending.ref, err) } return &OPFRuntimeFailedError{OPFCommand: redact.OPFCommand(), Cause: err} } } - // Pass 3: rebuild every commit before touching any ref, so a failure - // part-way through leaves every ref where it was. Each ref's chain is - // replayed ancestor→tip, so the rewritten parent carries into the next - // commit and the deepest one keeps the boundary parent. - rebuilt := make([]plumbing.Hash, len(pendings)) - for i, pr := range pendings { - parent := pr.base - for _, pc := range pr.commits { - redactedByPath := make(map[string][]byte, len(pc.blobs)) - for j, path := range pc.paths { - redactedByPath[path] = globalRedacted[pc.startIdx+j] - } - newHash, rebuildErr := rebuildCheckpointCommit(ctx, repo, pc.commit, parent, redactedByPath) - if rebuildErr != nil { - return fmt.Errorf("rebuild checkpoint commit %s on %s: %w", pc.commit.Hash.String()[:7], pr.ref, rebuildErr) - } - parent = newHash + // Replay ancestor→tip. A partial rebuild only creates unreachable objects; + // the ref moves once, after the whole replacement chain exists. + parent := pending.base + startIdx := 0 + for _, commit := range pending.commits { + redactedByPath := make(map[string][]byte, len(commit.blobs)) + for j, path := range commit.paths { + redactedByPath[path] = redacted[startIdx+j] } - rebuilt[i] = parent + startIdx += len(commit.blobs) + newHash, err := rebuildCheckpointCommit(ctx, repo, commit.commit, parent, redactedByPath) + if err != nil { + return fmt.Errorf("rebuild checkpoint commit %s on %s: %w", commit.commit.Hash.String()[:7], pending.ref, err) + } + parent = newHash } - // CAS each ref: a concurrent write that advanced a checkpoint ref during - // the rewrite must not be clobbered by our stale rebuild. - for i, pr := range pendings { - if err := checkpoint.CASPersistentRef(ctx, repo, pr.ref, rebuilt[i], pr.old); err != nil { - return fmt.Errorf("update checkpoint ref %s: %w", pr.ref, err) - } + // CAS prevents a concurrent checkpoint generation from being overwritten; + // the exact installed generation then replaces the old queue token. + return updateOPFRewrittenRef(ctx, repo, queue, pending.ref, parent, pending.old) +} + +type opfRefQueue interface { + EnqueueEntry(entry checkpoint.PushQueueEntry) error +} + +func updateOPFRewrittenRef( + ctx context.Context, + repo *git.Repository, + queue opfRefQueue, + refName plumbing.ReferenceName, + newHash, oldHash plumbing.Hash, +) error { + if err := checkpoint.CASPersistentRef(ctx, repo, refName, newHash, oldHash); err != nil { + return fmt.Errorf("update checkpoint ref %s: %w", refName, err) + } + if err := queue.EnqueueEntry(checkpoint.PushQueueEntry{Ref: refName, Hash: newHash}); err != nil { + return fmt.Errorf("enqueue rewritten checkpoint ref %s: %w", refName, err) } return nil } diff --git a/cmd/entire/cli/strategy/manual_commit_opf_rewrite.go b/cmd/entire/cli/strategy/manual_commit_opf_rewrite.go index e5cc6e9f3c..a3555aab6a 100644 --- a/cmd/entire/cli/strategy/manual_commit_opf_rewrite.go +++ b/cmd/entire/cli/strategy/manual_commit_opf_rewrite.go @@ -157,10 +157,11 @@ func resolveBootstrapLimit() int { return bootstrapDefaultLimit } -// OPFBatchTooLargeError: a single push has more prose-leaf content -// than ENTIRE_OPF_BATCH_LIMIT will allow OPF to chew through in one -// inference call. Pushing under the limit yields a single ~10-30s -// pause; without the cap, a 100MB-of-prose push could take an hour. +// OPFBatchTooLargeError: one OPF pass — a single checkpoint ref on +// git-refs, the whole unpushed v1 chain on git-branch — has more +// prose-leaf content than ENTIRE_OPF_BATCH_LIMIT will allow OPF to chew +// through in one inference call. Pushing under the limit yields a single +// ~10-30s pause; without the cap, a 100MB-of-prose push could take an hour. // // The user-facing remediation is identical in shape to // BootstrapTooLargeError: bump the limit, push without OPF, or break @@ -200,18 +201,18 @@ func resolveBatchLimit() int { return batchDefaultLimit } -// scaleBatchLimit multiplies a batch-limit value by mult, saturating -// at math.MaxInt to avoid signed-overflow when the limit is "unlimited" -// (math.MaxInt) or a very large explicit value. Returns 0 if either -// operand is non-positive, so the caller can treat 0 as "no cap" too. -func scaleBatchLimit(limit, mult int) int { - if limit <= 0 || mult <= 0 { +// rawByteCapForBatchLimit derives the raw-memory ceiling from the prose-leaf +// limit, saturating at math.MaxInt to avoid signed overflow when the limit is +// "unlimited" (math.MaxInt) or a very large explicit value. A non-positive +// limit returns 0 so the caller can treat it as "no cap" too. +func rawByteCapForBatchLimit(limit int) int { + if limit <= 0 { return 0 } - if limit > math.MaxInt/mult { + if limit > math.MaxInt/rawByteCapMultiplier { return math.MaxInt } - return limit * mult + return limit * rawByteCapMultiplier } // OPFRawBytesTooLargeError: the cumulative raw blob bytes the @@ -330,12 +331,12 @@ func RewriteUnpushedV1WithOPF(ctx context.Context, repo *git.Repository, target // Bound raw-bytes-in-memory incrementally so a pathological push // (e.g. 5 GiB of pasted dumps) aborts before exhausting the user's // shell RAM. The leaf-byte cap downstream is about inference cost; - // this one is about memory ceiling and fires earlier. scaleBatchLimit - // saturates at math.MaxInt so "unlimited" actually means unlimited - // — without saturation, "unlimited" × 100 overflows int and the cap - // trips on every push. - rawCap := scaleBatchLimit(resolveBatchLimit(), rawByteCapMultiplier) - var rawBytesSoFar int + // this one is about memory ceiling and fires earlier. + // rawByteCapForBatchLimit saturates at math.MaxInt so "unlimited" + // actually means unlimited — without saturation, "unlimited" × 100 + // overflows int and the cap trips on every push. + rawCap := rawByteCapForBatchLimit(resolveBatchLimit()) + rawBudget := newOPFRawByteBudget(rawCap) for _, c := range unpushed { pc := pendingCommit{commit: c} if !trailers.HasOPFApplied(c.Message) { @@ -350,15 +351,9 @@ func RewriteUnpushedV1WithOPF(ctx context.Context, repo *git.Repository, target // commit keeps the final rewritten tip from reintroducing an // un-OPF-redacted older shard. collect and apply walk the tree the // same way so the cached keys line up. - if err := collectTreeBlobs(repo, tree, "", &pc.blobs, &pc.paths); err != nil { + if err := collectTreeBlobsWithinBudget(repo, tree, "", &pc.blobs, &pc.paths, rawBudget); err != nil { return plumbing.ZeroHash, fmt.Errorf("collect blobs %s: %w", c.Hash.String()[:7], err) } - for _, b := range pc.blobs { - rawBytesSoFar += len(b.Content) - } - if rawBytesSoFar > rawCap { - return plumbing.ZeroHash, &OPFRawBytesTooLargeError{RawBytes: rawBytesSoFar, Limit: rawCap} - } globalBlobs = append(globalBlobs, pc.blobs...) } pendings = append(pendings, pc) @@ -369,6 +364,16 @@ func RewriteUnpushedV1WithOPF(ctx context.Context, repo *git.Repository, target // so a too-large push fails fast with a clear remediation message. var globalRedacted [][]byte if len(globalBlobs) > 0 { + // Unlike the git-refs backend (manual_commit_opf_refs.go), this cap + // stays cumulative across the WHOLE unpushed v1 chain rather than being + // scoped per commit: each rebuilt commit is the next one's parent (see + // the Pass 3 loop below), so commits cannot be redacted independently + // without reintroducing the chunking hazard: a partially-rewritten + // chain whose ancestor is a still-un-trailered commit. Separate checkpoint refs are independent chains, which is why + // only that backend can scope per ref. So if this cap ever trips on + // content that really is a chain of real sessions, this backend's only + // remaining levers are ENTIRE_OPF_BATCH_LIMIT or converging it onto the + // git-refs queue model — not further chunking. leafBytes := redact.SumProseLeafBytes(globalBlobs) if limit := resolveBatchLimit(); leafBytes > limit { return plumbing.ZeroHash, &OPFBatchTooLargeError{LeafBytes: leafBytes, Limit: limit} @@ -624,6 +629,29 @@ func isRedactableBlobName(name string) bool { // the tree (e.g. "ab/cd.../0/full.jsonl"), used later by the apply // walker to find each blob's redacted bytes in the cached map. func collectTreeBlobs(repo *git.Repository, tree *object.Tree, pathPrefix string, blobs *[]redact.NamedBlob, paths *[]string) error { + return collectTreeBlobsWithinBudget(repo, tree, pathPrefix, blobs, paths, nil) +} + +type opfRawByteBudget struct { + used int + limit int +} + +func newOPFRawByteBudget(limit int) *opfRawByteBudget { + return &opfRawByteBudget{limit: limit} +} + +// collectTreeBlobsWithinBudget checks each blob's object metadata before +// opening its content reader. That ordering makes the raw cap an allocation +// boundary rather than merely an after-the-fact input-size check. +func collectTreeBlobsWithinBudget( + repo *git.Repository, + tree *object.Tree, + pathPrefix string, + blobs *[]redact.NamedBlob, + paths *[]string, + budget *opfRawByteBudget, +) error { for _, e := range tree.Entries { switch e.Mode { case filemode.Dir: @@ -642,14 +670,14 @@ func collectTreeBlobs(repo *git.Repository, tree *object.Tree, pathPrefix string if err != nil { return fmt.Errorf("load subtree %s/%s: %w", pathPrefix, e.Name, err) } - if err := collectTreeBlobs(repo, subTree, subPath, blobs, paths); err != nil { + if err := collectTreeBlobsWithinBudget(repo, subTree, subPath, blobs, paths, budget); err != nil { return err } case filemode.Regular, filemode.Executable: if !isRedactableBlobName(e.Name) { continue } - content, err := readBlob(repo, e.Hash) + content, err := readBlobWithinBudget(repo, e.Hash, budget) if err != nil { return fmt.Errorf("read blob %s/%s: %w", pathPrefix, e.Name, err) } @@ -666,6 +694,60 @@ func collectTreeBlobs(repo *git.Repository, tree *object.Tree, pathPrefix string return nil } +func readBlobWithinBudget(repo *git.Repository, hash plumbing.Hash, budget *opfRawByteBudget) ([]byte, error) { + blob, err := repo.BlobObject(hash) + if err != nil { + return nil, fmt.Errorf("blob: %w", err) + } + if budget == nil { + return readBlobObject(blob, math.MaxInt) + } + + remaining := budget.limit - budget.used + if remaining < 0 || blob.Size > int64(remaining) { + rawBytes := budget.limit + if budget.limit < math.MaxInt { + rawBytes++ + } + if blob.Size <= int64(math.MaxInt-budget.used) { + rawBytes = budget.used + int(blob.Size) + } + return nil, &OPFRawBytesTooLargeError{RawBytes: rawBytes, Limit: budget.limit} + } + + content, err := readBlobObject(blob, remaining) + if err != nil { + var tooLarge *OPFRawBytesTooLargeError + if errors.As(err, &tooLarge) { + return nil, &OPFRawBytesTooLargeError{RawBytes: budget.used + remaining + 1, Limit: budget.limit} + } + return nil, err + } + budget.used += len(content) + return content, nil +} + +func readBlobObject(blob *object.Blob, remaining int) ([]byte, error) { + r, err := blob.Reader() + if err != nil { + return nil, fmt.Errorf("blob reader: %w", err) + } + defer func() { _ = r.Close() }() + + var reader io.Reader = r + if remaining < math.MaxInt { + reader = io.LimitReader(r, int64(remaining)+1) + } + data, err := io.ReadAll(reader) + if err != nil { + return nil, fmt.Errorf("blob read: %w", err) + } + if len(data) > remaining { + return nil, &OPFRawBytesTooLargeError{RawBytes: remaining + 1, Limit: remaining} + } + return data, nil +} + // rebuildTreeWithCachedRedaction walks the whole tree and produces a new // tree using the precomputed redactedByPath map for redactable blobs. // content_hash.txt files are recomputed in a second pass against the @@ -792,16 +874,7 @@ func readBlob(repo *git.Repository, hash plumbing.Hash) ([]byte, error) { if err != nil { return nil, fmt.Errorf("blob: %w", err) } - r, err := blob.Reader() - if err != nil { - return nil, fmt.Errorf("blob reader: %w", err) - } - defer func() { _ = r.Close() }() - data, err := io.ReadAll(r) - if err != nil { - return nil, fmt.Errorf("blob read: %w", err) - } - return data, nil + return readBlobObject(blob, math.MaxInt) } // atomicSetV1Ref CAS-updates the local v1 ref through the same lock protocol as diff --git a/cmd/entire/cli/strategy/manual_commit_opf_rewrite_test.go b/cmd/entire/cli/strategy/manual_commit_opf_rewrite_test.go index 66ca818a07..87e13e5e05 100644 --- a/cmd/entire/cli/strategy/manual_commit_opf_rewrite_test.go +++ b/cmd/entire/cli/strategy/manual_commit_opf_rewrite_test.go @@ -8,6 +8,7 @@ import ( "os" "os/exec" "path/filepath" + "strconv" "strings" "sync" "testing" @@ -89,6 +90,33 @@ func (f *fakeRuntimeAlwaysFails) RedactBatch(_ context.Context, _ []string, _ [] return nil, errors.New("simulated OPF runtime failure") } +// fakeRuntimeFailsOnSentinel fails Redact/RedactBatch only when a given text +// contains sentinel; otherwise it behaves like fakeOPFForRewrite (tags +// "PERSONABC" occurrences via findSentinelSpans). It lets a test make exactly +// one checkpoint ref's OPF call fail while its siblings' calls succeed — +// unlike fakeRuntimeAlwaysFails, which breaks the runtime for every ref. +type fakeRuntimeFailsOnSentinel struct { + sentinel string +} + +func (f *fakeRuntimeFailsOnSentinel) Redact(_ context.Context, text string, _ []string) ([]redact.Span, error) { + if strings.Contains(text, f.sentinel) { + return nil, errors.New("fake OPF failure on sentinel text") + } + return findSentinelSpans(text), nil +} + +func (f *fakeRuntimeFailsOnSentinel) RedactBatch(_ context.Context, inputs []string, _ []string) ([][]redact.Span, error) { + out := make([][]redact.Span, len(inputs)) + for i, in := range inputs { + if strings.Contains(in, f.sentinel) { + return nil, errors.New("fake OPF failure on sentinel text") + } + out[i] = findSentinelSpans(in) + } + return out, nil +} + // testOPFRuntime is the structural interface the redact package's // ConfigurePrivacyFilterWithRuntime accepts. Mirrors redact.opfRuntime // (unexported, can't be named directly from this package). @@ -583,15 +611,27 @@ func TestRewriteUnpushedV1WithOPF_BatchCap(t *testing.T) { envLimit string wantErr bool }{ - {name: "over_limit_rejected", envLimit: "10", wantErr: true}, + // 8000 is under the fixture's ~10 KB of prose-leaf content, and the + // RAM ceiling that scales off the same env var stays far above its + // ~13 KB of raw bytes, so it is the leaf-byte cap that trips. + {name: "over_limit_rejected", envLimit: "8000", wantErr: true}, {name: "unlimited_allows_any_size", envLimit: "unlimited", wantErr: false}, - {name: "env_override_allows_above_default", envLimit: "1000000", wantErr: false}, + // Above batchDefaultLimit, so the override is doing something. + {name: "env_override_allows_above_default", envLimit: "200000000", wantErr: false}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { configureFakeOPF(t, &fakeOPFForRewrite{}) t.Setenv(batchEnvVar, tc.envLimit) - repo, originalTip := setupV1Repo(t) // ~50 bytes of prose-leaf content + repo, _ := setupV1Repo(t) + // A prose-heavy second checkpoint (~10 KB of leaves, ~13 KB raw) + // so this fixture's raw-to-leaf ratio is ~1.2x, like real + // transcript content, rather than the ~22x of the tiny base + // fixture where structural JSON dominates. The caps are only + // meaningfully ordered against realistic content. + originalTip := addV1Checkpoint(t, repo, "b1b2c3d4e5f7", "prose-session", + strings.Repeat("the quick brown fox jumps over PERSONABC again ", 200), + strings.Repeat("look up PERSONABC please ", 40)) newTip, err := RewriteUnpushedV1WithOPF(context.Background(), repo, "origin") if !tc.wantErr { @@ -644,8 +684,9 @@ func TestOPFBatchTooLargeErrorMessage(t *testing.T) { // setupV1Repo fixture triggers it. func TestRewriteUnpushedV1WithOPF_RawByteCap(t *testing.T) { configureFakeOPF(t, &fakeOPFForRewrite{}) - // leaf cap = 1 → raw ceiling = 100 bytes; the setupV1Repo - // checkpoint writes far more than that across its shard blobs. + // leaf cap = 1 → raw ceiling = rawByteCapMultiplier bytes; the + // setupV1Repo checkpoint writes far more than that across its + // shard blobs. t.Setenv(batchEnvVar, "1") repo, originalTip := setupV1Repo(t) @@ -712,6 +753,33 @@ func TestCollectTreeBlobs_RedactsAllFileTypes(t *testing.T) { require.False(t, collectedNames[paths.ContentHashFileName], "content_hash.txt must be excluded from collection (deferred path)") } +func TestCollectTreeBlobsWithinBudget_RejectsBlobFromMetadataBeforeReadingContent(t *testing.T) { + tempDir := t.TempDir() + testutil.InitRepo(t, tempDir) + repo, err := git.PlainOpen(tempDir) + require.NoError(t, err) + + obj := repo.Storer.NewEncodedObject() + obj.SetType(plumbing.BlobObject) + w, err := obj.Writer() + require.NoError(t, err) + _, err = w.Write([]byte("larger than the configured budget")) + require.NoError(t, err) + require.NoError(t, w.Close()) + hash, err := repo.Storer.SetEncodedObject(obj) + require.NoError(t, err) + tree := &object.Tree{Entries: []object.TreeEntry{{Name: "transcript", Mode: filemode.Regular, Hash: hash}}} + + var blobs []redact.NamedBlob + var blobPaths []string + err = collectTreeBlobsWithinBudget(repo, tree, "", &blobs, &blobPaths, newOPFRawByteBudget(4)) + var tooLarge *OPFRawBytesTooLargeError + require.ErrorAs(t, err, &tooLarge) + assert.Greater(t, tooLarge.RawBytes, tooLarge.Limit) + assert.Empty(t, blobs, "an oversized blob must be rejected before its content is retained") + assert.Empty(t, blobPaths) +} + // The OPF rewrite must not touch externalized image assets: byte-redacting the // raw image blobs would corrupt them (breaking restore), and the fail-closed // rebuild would abort the push if they were collected but not redacted. Both @@ -857,6 +925,14 @@ func setupGitRefsOPFRepo(t *testing.T, cpIDs ...string) (bareDir string, repo *g // twice for the same checkpoint, it leaves the ref with two unpushed commits — // the shape a write followed by a backfill produces. func addGitRefsSession(t *testing.T, repo *git.Repository, cpID, sessionID string) { + t.Helper() + addGitRefsSessionWithTranscript(t, repo, cpID, sessionID, "Hello, PERSONABC asked") +} + +// addGitRefsSessionWithTranscript is addGitRefsSession with caller-chosen +// transcript text, so a test can make one checkpoint ref carry far more +// prose-leaf content than another. +func addGitRefsSessionWithTranscript(t *testing.T, repo *git.Repository, cpID, sessionID, transcript string) { t.Helper() stores, err := checkpoint.Open(t.Context(), repo, checkpoint.OpenOptions{}) require.NoError(t, err) @@ -865,7 +941,7 @@ func addGitRefsSession(t *testing.T, repo *git.Repository, cpID, sessionID strin CheckpointID: cid, SessionID: sessionID, Strategy: "manual-commit", - Transcript: redact.AlreadyRedacted([]byte(`{"role":"user","content":"Hello, PERSONABC asked"}` + "\n")), + Transcript: redact.AlreadyRedacted([]byte(fmt.Sprintf(`{"role":"user","content":%q}`+"\n", transcript))), Prompts: []string{"Look up PERSONABC"}, AuthorName: "Test", AuthorEmail: "test@test.com", @@ -901,6 +977,26 @@ func refHashes(t *testing.T, repo *git.Repository, refs []plumbing.ReferenceName return out } +func refRewriteSizes(t *testing.T, repo *git.Repository, refName plumbing.ReferenceName) (rawBytes, leafBytes int) { + t.Helper() + ref, err := repo.Reference(refName, true) + require.NoError(t, err) + chain, _, err := unappliedAncestry(repo, ref.Hash()) + require.NoError(t, err) + for _, commit := range chain { + tree, treeErr := repo.TreeObject(commit.TreeHash) + require.NoError(t, treeErr) + var blobs []redact.NamedBlob + var paths []string + require.NoError(t, collectTreeBlobs(repo, tree, "", &blobs, &paths)) + for _, blob := range blobs { + rawBytes += len(blob.Content) + } + leafBytes += redact.SumProseLeafBytes(blobs) + } + return rawBytes, leafBytes +} + // treeContents concatenates every file in the commit's tree. func treeContents(t *testing.T, repo *git.Repository, hash plumbing.Hash) string { t.Helper() @@ -929,8 +1025,22 @@ func queuedRefs(t *testing.T, repo *git.Repository) []plumbing.ReferenceName { return got } +func queuedRefEntries(t *testing.T, repo *git.Repository) []checkpoint.PushQueueEntry { + t.Helper() + queue, err := checkpoint.PushQueueForRepo(t.Context(), repo) + require.NoError(t, err) + entries, err := queue.PeekEntries() + require.NoError(t, err) + return entries +} + // Queued checkpoint refs are OPF-rewritten and stamped applied; a second run is // a no-op because the trailer marks them done (no re-scan, no ref movement). +// +// Scanning is scoped per ref — one OPF call per ref rather than one for the +// whole flush — so an oversized ref's cap failure cannot take its siblings down +// with it (see TestRewriteQueuedCheckpointRefsWithOPF_OversizedRefDoesNotBlockOthers). +// The cost of that isolation is N shell-outs instead of 1. func TestRewriteQueuedCheckpointRefsWithOPF_RewritesThenIsIdempotent(t *testing.T) { fake := &fakeOPFForRewrite{} configureFakeOPF(t, fake) @@ -938,10 +1048,14 @@ func TestRewriteQueuedCheckpointRefsWithOPF_RewritesThenIsIdempotent(t *testing. before := refHashes(t, repo, refs) require.NoError(t, RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo)) - require.Equal(t, 1, fake.batchCallCount(), "one OPF call for the whole flush") + require.Equal(t, 2, fake.batchCallCount(), "one OPF call per ref") after := refHashes(t, repo, refs) + entries := queuedRefEntries(t, repo) + require.Len(t, entries, len(refs)) for i, ref := range refs { + require.Equal(t, checkpoint.PushQueueEntry{Ref: ref, Hash: after[i]}, entries[i], + "the rewritten generation must replace the queue token for the old tip") require.NotEqual(t, before[i], after[i], "ref %s should have been rewritten", ref) commit, err := repo.CommitObject(after[i]) require.NoError(t, err) @@ -950,12 +1064,25 @@ func TestRewriteQueuedCheckpointRefsWithOPF_RewritesThenIsIdempotent(t *testing. } require.NoError(t, RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo)) - require.Equal(t, 1, fake.batchCallCount(), "already-applied refs must not be re-scanned") + require.Equal(t, 2, fake.batchCallCount(), "already-applied refs must not be re-scanned") require.Equal(t, after, refHashes(t, repo, refs), "already-applied refs must keep their exact hash") } -// Fail-closed: an OPF failure anywhere in the batch must leave EVERY queued ref -// where it was, so no ref is stamped applied over content OPF never saw. +// Fail-closed: an OPF *runtime* failure must leave EVERY queued ref where it +// was, so no ref is stamped applied over content OPF never saw. +// +// This survives per-ref scoping deliberately, and is the reason the per-ref loop +// re-checks the circuit breaker on every iteration. The first ref's failure +// trips the process-wide breaker, and a tripped breaker makes +// BatchBytesWithPrivacyFilter return regex-only output with a *nil* error — so +// a loop that merely skipped the failing ref and carried on would rebuild every +// later ref from regex-only content and stamp Entire-OPF-Applied on it. A +// broken runtime is broken for the whole process; only the size cap is a +// per-ref condition. +// +// Case (a) of the pair: the FIRST ref's call fails, so nothing is rewritten. +// TestRewriteQueuedCheckpointRefsWithOPF_FailingOPFRefDoesNotBlockOthers is +// case (b) — a ref already rewritten before the failure keeps its rewrite. func TestRewriteQueuedCheckpointRefsWithOPF_FailureLeavesEveryRefUnmoved(t *testing.T) { configureFakeOPF(t, &fakeRuntimeAlwaysFails{}) _, repo, refs := setupGitRefsOPFRepo(t, "a1b2c3d4e5f6", "b2c3d4e5f6a1") @@ -965,6 +1092,226 @@ func TestRewriteQueuedCheckpointRefsWithOPF_FailureLeavesEveryRefUnmoved(t *test var runtimeFail *OPFRuntimeFailedError require.ErrorAs(t, err, &runtimeFail) require.Equal(t, before, refHashes(t, repo, refs)) + for _, ref := range refHashes(t, repo, refs) { + commit, cErr := repo.CommitObject(ref) + require.NoError(t, cErr) + require.False(t, trailers.HasOPFApplied(commit.Message), + "no ref may be stamped OPF-applied once the runtime is broken") + } +} + +// Per-ref runtime-error scoping: a ref whose own OPF call fails is left +// unmoved, while a ref already rewritten earlier in the same flush keeps its +// rewrite instead of being discarded with it. Complements +// _OversizedRefDoesNotBlockOthers, which covers the cap-failure path only: this +// one pins a genuine BatchBytesWithPrivacyFilter error, asserted through the +// error's Cause so a breaker-skip can't pass for a real per-ref call. +// +// Ref ORDER here is load-bearing, not incidental. redact.handleOPFFailure trips +// the process-wide breaker on the FIRST runtime failure (a CompareAndSwap, not +// an N-strike counter), and a tripped breaker makes BatchBytesWithPrivacyFilter +// return regex-only content with a NIL error — so the top-of-loop breaker +// check deliberately breaks out of the whole loop rather than continuing to the +// next ref. There is therefore no ref that can be OPF-scanned *after* a runtime +// failure, and a test that ordered the failing ref first would pass for the +// wrong reason (the sibling skipped by the breaker-break, not isolated by the +// per-ref logic — that is what _FailureLeavesEveryRefUnmoved pins). What per-ref +// scoping buys under a runtime error is exactly what this test asserts: the ref +// processed before the failure keeps the rewrite OPF really did scan. +func TestRewriteQueuedCheckpointRefsWithOPF_FailingOPFRefDoesNotBlockOthers(t *testing.T) { + const okID, failID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + const failSentinel = "OPFBOOM" + configureFakeOPF(t, &fakeRuntimeFailsOnSentinel{sentinel: failSentinel}) + // The push queue preserves first-seen order, so the refs are processed in + // the order they are written here: the surviving ref first, the failing + // one second. + _, repo, refs := setupGitRefsOPFRepo(t, okID, failID) + addGitRefsSessionWithTranscript(t, repo, failID, "sess-fail", + "Hello, PERSONABC asked about "+failSentinel) + before := refHashes(t, repo, refs) + + err := RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo) + + var runtimeFail *OPFRuntimeFailedError + require.ErrorAs(t, err, &runtimeFail, "the failing ref must still report its own runtime failure") + require.ErrorContains(t, runtimeFail.Cause, "fake OPF failure on sentinel text", + "the error must carry the failing ref's own OPF call as its cause, not the bare breaker check") + + after := refHashes(t, repo, refs) + require.NotEqual(t, before[0], after[0], + "the ref whose OPF call succeeded must be rewritten despite its failing sibling") + survivor, cErr := repo.CommitObject(after[0]) + require.NoError(t, cErr) + require.True(t, trailers.HasOPFApplied(survivor.Message)) + require.NotContains(t, treeContents(t, repo, after[0]), "PERSONABC") + + require.Equal(t, before[1], after[1], "the failing ref must be left exactly where it was") + unscanned, cErr := repo.CommitObject(after[1]) + require.NoError(t, cErr) + require.False(t, trailers.HasOPFApplied(unscanned.Message), + "a ref OPF never finished scanning must not be stamped applied") + require.Contains(t, treeContents(t, repo, after[1]), "PERSONABC", + "the failing ref's content must be left byte-identical, sentinel included") +} + +// The raw-byte memory ceiling is scoped to the one ref whose blobs are resident +// for the current OPF call. Two refs that each fit must both rewrite even when +// their combined raw bytes would exceed one ref's ceiling. +func TestRewriteQueuedCheckpointRefsWithOPF_RawByteCapIsPerRef(t *testing.T) { + fake := &fakeOPFForRewrite{} + configureFakeOPF(t, fake) + const firstID, secondID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + _, repo, refs := setupGitRefsOPFRepo(t, firstID, secondID) + + // Pad both refs with raw bytes that are not prose leaves (no spaces), so + // each ref's raw size approaches the ceiling its leaf content allows + // without moving the leaf-byte cap. + _, leaf0 := refRewriteSizes(t, repo, refs[0]) + _, leaf1 := refRewriteSizes(t, repo, refs[1]) + padding := strings.Repeat("x", max(leaf0, leaf1)*rawByteCapMultiplier*6/10) + addGitRefsSessionWithTranscript(t, repo, firstID, "sess-pad-0", padding) + addGitRefsSessionWithTranscript(t, repo, secondID, "sess-pad-1", padding) + + raw0, leaf0 := refRewriteSizes(t, repo, refs[0]) + raw1, leaf1 := refRewriteSizes(t, repo, refs[1]) + limit := max(max(leaf0, leaf1), (max(raw0, raw1)+rawByteCapMultiplier-1)/rawByteCapMultiplier) + rawCap := rawByteCapForBatchLimit(limit) + require.LessOrEqual(t, raw0, rawCap) + require.LessOrEqual(t, raw1, rawCap) + require.Greater(t, raw0+raw1, rawCap, + "fixture must exceed one raw ceiling only when refs are combined") + t.Setenv(batchEnvVar, strconv.Itoa(limit)) + before := refHashes(t, repo, refs) + + require.NoError(t, RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo)) + require.Equal(t, 2, fake.batchCallCount(), "each fitting ref must reach OPF") + after := refHashes(t, repo, refs) + for i := range refs { + require.NotEqual(t, before[i], after[i]) + commit, err := repo.CommitObject(after[i]) + require.NoError(t, err) + require.True(t, trailers.HasOPFApplied(commit.Message)) + } +} + +// A raw-oversized ref is a local failure. It stays untouched and reports the +// cap error, while a later fitting sibling is still collected, scanned, and +// rewritten. +func TestRewriteQueuedCheckpointRefsWithOPF_RawOversizedRefDoesNotBlockLaterRef(t *testing.T) { + const oversizedID, fitsID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + fake := &fakeOPFForRewrite{} + configureFakeOPF(t, fake) + _, repo, refs := setupGitRefsOPFRepo(t, oversizedID, fitsID) + addGitRefsSessionWithTranscript(t, repo, oversizedID, "sess-raw-oversized", + strings.Repeat("the quick brown fox jumps over PERSONABC again ", 800)) + + oversizedRaw, _ := refRewriteSizes(t, repo, refs[0]) + fitsRaw, fitsLeaf := refRewriteSizes(t, repo, refs[1]) + limit := max(fitsLeaf, (fitsRaw+rawByteCapMultiplier-1)/rawByteCapMultiplier) + rawCap := rawByteCapForBatchLimit(limit) + require.Greater(t, oversizedRaw, rawCap) + require.LessOrEqual(t, fitsRaw, rawCap) + t.Setenv(batchEnvVar, strconv.Itoa(limit)) + before := refHashes(t, repo, refs) + + err := RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo) + + var tooLarge *OPFRawBytesTooLargeError + require.ErrorAs(t, err, &tooLarge) + require.Greater(t, tooLarge.RawBytes, tooLarge.Limit) + after := refHashes(t, repo, refs) + require.Equal(t, before[0], after[0], "raw-oversized ref must remain byte-identical") + require.NotEqual(t, before[1], after[1], "later fitting ref must still rewrite") + require.Equal(t, 1, fake.batchCallCount(), "only the fitting ref may reach OPF") + fitting, commitErr := repo.CommitObject(after[1]) + require.NoError(t, commitErr) + require.True(t, trailers.HasOPFApplied(fitting.Message)) +} + +// A broken OPF runtime withholds every ref after it, so it outranks an earlier +// ref's cap error: reporting only the cap would send the user after the wrong +// problem. +func TestRewriteQueuedCheckpointRefsWithOPF_RuntimeFailureOutranksEarlierCapError(t *testing.T) { + const oversizedID, fitsID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + configureFakeOPF(t, &fakeRuntimeAlwaysFails{}) + _, repo, refs := setupGitRefsOPFRepo(t, oversizedID, fitsID) + addGitRefsSessionWithTranscript(t, repo, oversizedID, "sess-raw-oversized", + strings.Repeat("the quick brown fox jumps over PERSONABC again ", 800)) + + oversizedRaw, _ := refRewriteSizes(t, repo, refs[0]) + fitsRaw, fitsLeaf := refRewriteSizes(t, repo, refs[1]) + limit := max(fitsLeaf, (fitsRaw+rawByteCapMultiplier-1)/rawByteCapMultiplier) + require.Greater(t, oversizedRaw, rawByteCapForBatchLimit(limit)) + t.Setenv(batchEnvVar, strconv.Itoa(limit)) + before := refHashes(t, repo, refs) + + err := RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo) + + var runtimeErr *OPFRuntimeFailedError + require.ErrorAs(t, err, &runtimeErr, "the runtime failure must not be hidden behind the first ref's cap error") + assert.Equal(t, before, refHashes(t, repo, refs), "neither ref may move") +} + +// Queue order is stable across pushes, so a ref whose OPF call fails at +// runtime moves to the back of the queue: the next push reaches the refs +// behind it first instead of stopping on the same ref every time. +func TestRewriteQueuedCheckpointRefsWithOPF_RuntimeFailureRotatesTheFailedRef(t *testing.T) { + configureFakeOPF(t, &fakeRuntimeAlwaysFails{}) + _, repo, refs := setupGitRefsOPFRepo(t, "a1b2c3d4e5f6", "b2c3d4e5f6a1") + queue, err := checkpoint.PushQueueForRepo(t.Context(), repo) + require.NoError(t, err) + before, err := queue.Peek() + require.NoError(t, err) + require.Equal(t, refs, before) + + err = RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo) + + var runtimeErr *OPFRuntimeFailedError + require.ErrorAs(t, err, &runtimeErr) + after, err := queue.Peek() + require.NoError(t, err) + assert.Equal(t, []plumbing.ReferenceName{refs[1], refs[0]}, after, + "the ref that failed must move behind the ref it was withholding") +} + +// Per-ref cap scoping: the leaf-byte cap is enforced per ref, not across the +// whole flush, so one oversized ref no longer poisons the rewrite of every ref +// queued alongside it. The ref that fits is redacted, tagged, and moved; the +// oversized one is left byte-identical and stays queued, and only it is named +// in the returned error. +// +// This exercises the rewrite operation directly. Delivery independently checks +// each immutable candidate's trailer before deciding whether it may ship. +func TestRewriteQueuedCheckpointRefsWithOPF_OversizedRefDoesNotBlockOthers(t *testing.T) { + const fitsID, oversizedID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + fake := &fakeOPFForRewrite{} + configureFakeOPF(t, fake) + _, repo, refs := setupGitRefsOPFRepo(t, fitsID, oversizedID) + // A second, much larger session carries the oversized ref's own prose-leaf + // bytes (~10 KB) past the cap while the other ref stays well under it. The + // raw-byte ceiling (cap × rawByteCapMultiplier, so 16 KB here) stays above + // the flush's raw bytes, so the leaf-byte cap is what this test observes. + addGitRefsSessionWithTranscript(t, repo, oversizedID, "sess-oversized", + strings.Repeat("the quick brown fox jumps over PERSONABC again ", 200)) + t.Setenv(batchEnvVar, "8000") + before := refHashes(t, repo, refs) + + err := RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo) + + var tooLarge *OPFBatchTooLargeError + require.ErrorAs(t, err, &tooLarge, "the oversized ref must still report its own cap failure") + require.Greater(t, tooLarge.LeafBytes, tooLarge.Limit) + + after := refHashes(t, repo, refs) + require.NotEqual(t, before[0], after[0], + "the ref that fits must be rewritten despite its oversized sibling") + fitting, cErr := repo.CommitObject(after[0]) + require.NoError(t, cErr) + require.True(t, trailers.HasOPFApplied(fitting.Message)) + require.NotContains(t, treeContents(t, repo, after[0]), "PERSONABC") + + require.Equal(t, before[1], after[1], "the oversized ref must be left exactly where it was") + require.Equal(t, 1, fake.batchCallCount(), "only the ref that fits may reach OPF") } // Backend divergence: the v1 path aborts the user's push on OPF failure; the @@ -982,7 +1329,7 @@ func TestPrePushCheckpointRefs_OPFFailureWithholdsFlush(t *testing.T) { require.NoError(t, NewManualCommitStrategy().PrePushFromGitHook(t.Context(), "origin"), "an OPF failure must not block the user's git push") - assert.Contains(t, buf.String(), "checkpoint refs", "the withheld push must be visible to the user") + assert.Contains(t, buf.String(), "checkpoint ref", "the withheld push must be visible to the user") assert.ElementsMatch(t, refs, queuedRefs(t, repo), "withheld refs stay queued for the next push") lsCmd := exec.CommandContext(t.Context(), "git", "ls-remote", bareDir) lsCmd.Env = testutil.GitIsolatedEnv() @@ -991,6 +1338,56 @@ func TestPrePushCheckpointRefs_OPFFailureWithholdsFlush(t *testing.T) { assert.NotContains(t, string(out), refs[0].String(), "no ref may reach the remote when OPF failed") } +// cancelOPFPrompt makes the git-refs decision an explicit Ctrl-C at the prompt, +// which no environment variable or setting can express. +func cancelOPFPrompt(t *testing.T) { + t.Helper() + old := opfCheckpointRefsDecision + opfCheckpointRefsDecision = func(context.Context) (OPFDecision, error) { return OPFAbort, nil } + t.Cleanup(func() { opfCheckpointRefsDecision = old }) +} + +// Ctrl-C at the OPF prompt ships nothing, not even a ref an earlier push already +// redacted and tagged: cancel means the checkpoint refs stay queued. +func TestPrePushCheckpointRefs_OPFCancelShipsNothing(t *testing.T) { + configureFakeOPF(t, &fakeOPFForRewrite{}) + bareDir, repo, refs := setupGitRefsOPFRepo(t, "a1b2c3d4e5f6") + require.NoError(t, RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo)) + tagged := refHashes(t, repo, refs) + commit, err := repo.CommitObject(tagged[0]) + require.NoError(t, err) + require.True(t, trailers.HasOPFApplied(commit.Message), "fixture: the queued ref must already be deliverable") + cancelOPFPrompt(t) + + var buf bytes.Buffer + oldWriter := stderrWriter + stderrWriter = &buf + t.Cleanup(func() { stderrWriter = oldWriter }) + + require.NoError(t, NewManualCommitStrategy().PrePushFromGitHook(t.Context(), "origin"), + "cancelling OPF must not block the user's git push") + + assert.Contains(t, buf.String(), "1 checkpoint ref(s) were not pushed") + assert.Equal(t, tagged, refHashes(t, repo, refs)) + assert.ElementsMatch(t, refs, queuedRefs(t, repo), "cancelled refs stay queued") + assertRefsAbsentFromRemote(t, bareDir, refs, "a cancelled push must not reach the remote") +} + +func TestPushQueuedCheckpointRefs_OPFCancelShipsNothing(t *testing.T) { + configureFakeOPF(t, &fakeOPFForRewrite{}) + bareDir, repo, refs := setupGitRefsOPFRepo(t, "a1b2c3d4e5f6") + require.NoError(t, RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo)) + cancelOPFPrompt(t) + + pushed, pushDisabled, err := PushQueuedCheckpointRefs(t.Context(), repo, bareDir) + + require.ErrorIs(t, err, ErrOPFAbortedByUser) + assert.False(t, pushDisabled) + assert.Zero(t, pushed) + assert.ElementsMatch(t, refs, queuedRefs(t, repo), "cancelled refs stay queued") + assertRefsAbsentFromRemote(t, bareDir, refs, "a cancelled push must not reach the remote") +} + // With OPF off the git-refs path is unchanged: refs push as written, unrewritten. func TestPrePushCheckpointRefs_OPFDisabledFlushesUnchanged(t *testing.T) { redact.ResetOPFConfigForTest() diff --git a/cmd/entire/cli/strategy/manual_commit_push.go b/cmd/entire/cli/strategy/manual_commit_push.go index 038569eddb..4f08efbfea 100644 --- a/cmd/entire/cli/strategy/manual_commit_push.go +++ b/cmd/entire/cli/strategy/manual_commit_push.go @@ -17,6 +17,7 @@ import ( checkpointremote "github.com/entireio/cli/cmd/entire/cli/checkpoint/remote" "github.com/entireio/cli/cmd/entire/cli/logging" "github.com/entireio/cli/cmd/entire/cli/settings" + "github.com/entireio/cli/cmd/entire/cli/trailers" "github.com/entireio/cli/perf" "github.com/entireio/cli/redact" ) @@ -33,6 +34,10 @@ var ErrOPFAbortedByUser = errors.New("OPF prompt aborted by user; push cancelled var opfPrePushProgressWriter io.Writer = os.Stderr +// opfCheckpointRefsDecision is opfPrePushDecision for the git-refs paths; the +// prompt is the only way to an explicit abort otherwise. +var opfCheckpointRefsDecision = opfPrePushDecision //nolint:gochecknoglobals // explicit-abort test seam + // PrePush is called by the git pre-push hook before pushing to a remote. // It pushes each ref in refs.Push alongside the user's push. // @@ -245,74 +250,141 @@ func opfPrePushDecision(ctx context.Context) (OPFDecision, error) { return resolveOPFDecisionForPrePush(ctx, opfCfg, opfPrePushProgressWriter) } -// opfGateForCheckpointRefs resolves the OPF decision and, when it says run, -// rewrites the queued checkpoint refs. A nil error means the flush may proceed; -// a non-nil one means it must be withheld, leaving the refs queued rather than -// shipping 8-layer content the user did not opt out of. OPF being disabled is a -// nil error, so every flush can call this unconditionally. +// opfDecisionForCheckpointRefs resolves whether this push requires OPF. It does +// not run model inference: callers run the rewrite (rewriteCheckpointRefsForPush) +// only when the decision is OPFRun. +// +// The resolved decision is returned alongside the error because delivery needs +// it: OPFSkip leaves the whole queue un-rewritten and is indistinguishable from +// a failed rewrite by inspecting the refs, but it is the user choosing to ship +// 8-layer content for this push, so nothing may be withheld for lacking the +// trailer (see deliveryRequiresOPFTrailer). // // Every path that flushes the queue must pass through here, because skipping it // is not the same as an OPFSkip: a skip is a decision the user made for this -// push and it ships untagged content deliberately, while a missing gate ships -// exactly the same bytes having never asked. +// push and it ships untagged content deliberately, while a missing decision +// ships exactly the same bytes having never asked. // // Reporting belongs to the caller: the pre-push hook warns and carries on, // because a checkpoint failure must never block the user's own git push, while // an explicitly requested push surfaces the error instead. // -// Precondition, and the one way to make this gate a silent no-op: OPFEnabled +// Precondition, and the one way to make this decision a silent no-op: OPFEnabled // reads process-global config that only EnsureRedactionConfigured sets, so a // caller that reaches here without it having run reads "OPF off" and flushes // everything unscanned. Every caller must ensure it — the hook path via // withHookSession (hooks_git_cmd.go), the migration push via pushMigratedRefs // (doctor_migrate.go). // -// This gate does not call it itself: on the hook path its scanner-config error -// is deliberately logged and survived rather than propagated, so raising it -// here would withhold pushes on a condition that path documents as -// non-fatal. -func opfGateForCheckpointRefs(ctx context.Context, repo *git.Repository) error { +// This decision helper does not call it itself: on the hook path its +// scanner-config error is deliberately logged and survived rather than +// propagated, so raising it here would withhold pushes on a condition that path +// documents as non-fatal. +func opfDecisionForCheckpointRefs(ctx context.Context) (OPFDecision, error) { if !redact.OPFEnabled() { - return nil + return OPFSkip, nil } - decision, err := opfPrePushDecision(ctx) + decision, err := opfCheckpointRefsDecision(ctx) if err != nil { - return err + return OPFAbort, err } switch decision { case OPFAbort: - return ErrOPFAbortedByUser + return OPFAbort, ErrOPFAbortedByUser case OPFSkip: // Explicit opt-out for this push: flush the 8-layer content as-is, // untagged — same as the v1 path. logging.Info(ctx, "OPF skipped for this push (user choice or settings)") case OPFRun: - _, opfSpan := perf.Start(ctx, "opf_pre_push_rewrite_refs") - defer opfSpan.End() - if rewriteErr := RewriteQueuedCheckpointRefsWithOPF(ctx, repo); rewriteErr != nil { - opfSpan.RecordError(rewriteErr) - return rewriteErr - } + // The caller runs the rewrite after this decision. + } + return decision, nil +} + +func rewriteCheckpointRefsForPush(ctx context.Context, repo *git.Repository) error { + _, opfSpan := perf.Start(ctx, "opf_rewrite_refs") + defer opfSpan.End() + if err := RewriteQueuedCheckpointRefsWithOPF(ctx, repo); err != nil { + opfSpan.RecordError(err) + return err } return nil } -// warnOPFCheckpointRefsWithheld reports a withheld flush on both channels: the -// log for diagnosis, and the user's terminal so a silently un-synced checkpoint -// is never the first they hear of it. +// deliveryRequiresOPFTrailer reports whether this push may deliver only the +// queued refs that already carry the OPF trailer, given the decision +// opfDecisionForCheckpointRefs resolved. It is the whole of the fail-closed delivery +// contract: under every decision but OPFSkip a ref ships once it already carries +// the trailer, never because we gave up waiting for it. +// +// Deliberately a pure predicate over the decision and nothing else. The refs it +// applies to are read inside flushCheckpointRefsQueue, from the set that flush +// itself drained, so there is no window in which a ref enqueued by a concurrent +// session can be absent from a readiness snapshot yet present in the push. +// Answering "which refs?" here instead is what opened that window. +// +// A decision error is likewise not an input. Whether the decision was aborted +// or could not be resolved, the question "may this ref ship?" has the same +// answer, and the ref's own trailer answers it. +// +// OPFSkip is an explicit opt-out for this push (or OPF off entirely): nothing is +// withheld and the queue ships as-is, untagged. Returning false also keeps the +// OPF-off fast path free of the per-ref commit loads a trailer check would cost. +// +// OPFAbort reaches here only for a decision that could not be resolved, and is +// treated as OPFRun: a ref that already carries the trailer is no less redacted +// for the decision having failed. An explicit cancel never gets this far; the +// callers withhold the whole flush for ErrOPFAbortedByUser (see +// opfCancelledCheckpointRefs). +func deliveryRequiresOPFTrailer(decision OPFDecision) bool { + return decision != OPFSkip +} + +// warnOPFCheckpointRefsWithheld reports refs the delivery check held back on both +// channels: the log for diagnosis, and the user's terminal so a silently +// un-synced checkpoint is never the first they hear of it. withheld is how many +// queued refs this push is leaving behind — zero means the decision failure +// cost this push nothing to deliver (every queued ref already carried the +// trailer), so there is nothing to tell the user to look for. // // Names no cause, for the reason PushQueuedCheckpointRefs does not either: the -// gate withholds on an unresolvable decision or a failed scan, but also on a -// ref update that lost a race after a scan that ran fine. Saying "OPF did not -// run" on the pre-push channel — the one nearly every withheld flush comes -// through — would send those users after the wrong problem. The wrapped error -// says which it was. -func warnOPFCheckpointRefsWithheld(ctx context.Context, err error) { +// delivery check can withhold after an unresolvable decision, or because the +// rewrite could not finish a ref. Saying "OPF did not run" on the pre-push +// channel would send some users after the wrong problem. The wrapped error says +// which decision failure occurred when one exists. +func warnOPFCheckpointRefsWithheld(ctx context.Context, err error, withheld int) { logging.Warn(ctx, "checkpoint ref push withheld; refs left queued", slog.String("error", err.Error()), + slog.Int("withheld", withheld), ) + if withheld == 0 { + return + } fmt.Fprintf(stderrWriter, - "[entire] Your checkpoint refs were not pushed and stay queued for the next push: %v\n", err) + "[entire] %d checkpoint ref(s) were not pushed and stay queued for the next push: %v\n", withheld, err) +} + +// opfCancelledCheckpointRefs reports whether the user explicitly cancelled OPF +// at the prompt. That withholds the whole flush, already-trailered refs +// included: cancel means nothing ships, and the docs promise the refs stay +// queued. +func opfCancelledCheckpointRefs(opfErr error) bool { + return errors.Is(opfErr, ErrOPFAbortedByUser) +} + +// queuedCheckpointRefCount is how many refs a withheld flush leaves queued, for +// the user-facing warning. A queue that cannot be read counts as zero, which +// keeps the warning in the log only. +func queuedCheckpointRefCount(ctx context.Context, repo *git.Repository) int { + queue, err := checkpoint.PushQueueForRepo(ctx, repo) + if err != nil { + return 0 + } + entries, err := queue.PeekEntries() + if err != nil { + return 0 + } + return len(entries) } // deferCheckpointPushOnEmptyRemote reports whether publication of the git-branch @@ -409,8 +481,9 @@ func remoteHasTrackingRefs(ctx context.Context, remote string) bool { // recorded refs fast-forward-only (git-refs primary; never a force push — a // diverged ref is recovered via fetch+replay). Transient push failures are logged and // swallowed — like the v1 path, they must not block the user's git push — and the -// refs stay queued for the next pre-push. When OPF is enabled the queued refs -// are re-redacted with it first (RewriteQueuedCheckpointRefsWithOPF). +// refs stay queued for the next pre-push. When OPF is enabled, each queued ref +// is rewritten in turn and only immutable generations that carry the OPF +// trailer ship; a ref OPF could not finish stays queued. func (s *ManualCommitStrategy) prePushCheckpointRefs(ctx context.Context, ps pushSettings, pendingCapture string) error { repo, err := OpenRepository(ctx) if err != nil { @@ -423,14 +496,29 @@ func (s *ManualCommitStrategy) prePushCheckpointRefs(ctx context.Context, ps pus // OPF backend divergence: both paths fail closed, but this one does it // without blocking the user. The v1 path aborts the user's git push; here a // checkpoint-ref failure must never do that (see this function's doc), so - // failing closed means withholding the flush — nothing un-OPF'd ships, the - // refs stay queued, and the user's push proceeds. - if opfErr := opfGateForCheckpointRefs(ctx, repo); opfErr != nil { - warnOPFCheckpointRefsWithheld(ctx, opfErr) + // failing closed means withholding the refs OPF has not finished with — + // nothing un-OPF'd ships, those refs stay queued, and the user's push + // proceeds. + opfDecision, opfErr := opfDecisionForCheckpointRefs(ctx) + if opfCancelledCheckpointRefs(opfErr) { + warnOPFCheckpointRefsWithheld(ctx, opfErr, queuedCheckpointRefCount(ctx, repo)) return nil } - - if flushed, err := flushCheckpointRefsQueue(ctx, repo, ps); err == nil { + if opfDecision == OPFRun && opfErr == nil { + // Each queued ref is rewritten and capped on its own, so a ref OPF + // cannot finish stays untrailered while its siblings are rewritten. + // Delivery below reads each ref's own tip trailer, and exact queue-token + // cleanup preserves any generation that advanced during this push. + opfErr = rewriteCheckpointRefsForPush(ctx, repo) + } + flushed, withheld, flushErr := flushCheckpointRefsQueue(ctx, repo, ps, deliveryRequiresOPFTrailer(opfDecision)) + if opfErr != nil { + // After the flush, because the count of refs left behind is the flush's + // answer to give: it is the one that read each ref's trailer against the + // set it drained. + warnOPFCheckpointRefsWithheld(ctx, opfErr, withheld) + } + if flushErr == nil { // Delivered, and only if something actually was: an empty queue pushed // nothing, so it must not move the election or announce that it had. if pendingCapture != "" && flushed > 0 { @@ -445,7 +533,7 @@ func (s *ManualCommitStrategy) prePushCheckpointRefs(ctx context.Context, ps pus // Fail-soft: a checkpoint-ref push failure must never block the user's // git push. The refs stay queued for the next pre-push. logging.Warn(ctx, "git-refs pre-push: checkpoint ref push failed; refs left queued", - slog.String("error", err.Error())) + slog.String("error", flushErr.Error())) } return nil } @@ -455,61 +543,139 @@ func (s *ManualCommitStrategy) prePushCheckpointRefs(ctx context.Context, ps pus // caller owns the repo. It returns the number of refs pushed and whether // pushing is disabled in settings — a distinct signal from pushed==0 with // pushing enabled (an empty queue), so callers can report the two accurately. -// Like the pre-push paths, an OPF rewrite that cannot run when OPF is enabled -// errors with the refs left queued. Currently used by the checkpoint migration +// Like the pre-push paths, a ref OPF has not finished with is left queued +// rather than shipped, and the leftover is reported as an error once the refs +// that were ready have gone. Currently used by the checkpoint migration // command's opt-in "push now". +// +// pushed can therefore be non-zero alongside a non-nil err: callers must report +// what landed rather than reading an error as "nothing happened". func PushQueuedCheckpointRefs(ctx context.Context, repo *git.Repository, remote string) (pushed int, pushDisabled bool, err error) { ps := resolvePushSettings(ctx, remote) if ps.pushDisabled { return 0, true, nil } - if opfErr := opfGateForCheckpointRefs(ctx, repo); opfErr != nil { + opfDecision, opfErr := opfDecisionForCheckpointRefs(ctx) + if opfCancelledCheckpointRefs(opfErr) { + return 0, false, opfErr + } + if opfDecision == OPFRun && opfErr == nil { + opfErr = rewriteCheckpointRefsForPush(ctx, repo) + } + var withheld int + pushed, withheld, err = flushCheckpointRefsQueue(ctx, repo, ps, deliveryRequiresOPFTrailer(opfDecision)) + if err != nil { + return pushed, false, err + } + if opfErr != nil && withheld > 0 { // Names no cause, matching flushCheckpointRefsQueue's retry message - // below: the gate fails on an unresolvable decision or a failed scan, - // but also on a ref update that lost a race after a scan that ran - // fine, so "OPF did not run" would sometimes send the user after the - // wrong problem. The wrapped error says which it was. - return 0, false, fmt.Errorf("checkpoint refs stay queued: %w", opfErr) - } - pushed, err = flushCheckpointRefsQueue(ctx, repo, ps) - return pushed, false, err + // below: the decision can fail before a rewrite, or the synchronous + // rewrite can fail on one ref after others completed. The wrapped error + // says which it was. pushed already reflects whatever shipped — this + // reports that some refs remain queued, not that nothing did. An error + // that withheld nothing is not reported: there is no leftover to act on. + return pushed, false, fmt.Errorf("%d checkpoint ref(s) stay queued: %w", withheld, opfErr) + } + return pushed, false, nil +} + +// partitionCheckpointRefPushes captures each queued ref's current immutable tip +// and, when required, verifies the trailer on that exact hash. The queue token +// remains attached solely for exact-generation cleanup after delivery. +func partitionCheckpointRefPushes( + repo *git.Repository, + entries []checkpoint.PushQueueEntry, + requireOPFTrailer bool, +) (ready []checkpointRefPush, awaiting, stale []checkpoint.PushQueueEntry) { + ready = make([]checkpointRefPush, 0, len(entries)) + for _, entry := range entries { + ref, err := repo.Reference(entry.Ref, true) + if errors.Is(err, plumbing.ErrReferenceNotFound) { + stale = append(stale, entry) + continue + } + if err != nil { + awaiting = append(awaiting, entry) + continue + } + candidate := checkpointRefPush{token: entry, name: entry.Ref, hash: ref.Hash()} + if requireOPFTrailer { + commit, commitErr := repo.CommitObject(candidate.hash) + if commitErr != nil || !trailers.HasOPFApplied(commit.Message) { + awaiting = append(awaiting, entry) + continue + } + } + ready = append(ready, candidate) + } + return ready, awaiting, stale +} + +func checkpointRefPushTokens(refs []checkpointRefPush) []checkpoint.PushQueueEntry { + entries := make([]checkpoint.PushQueueEntry, 0, len(refs)) + for _, ref := range refs { + entries = append(entries, ref.token) + } + return entries } // flushCheckpointRefsQueue drains the push-discovery queue and batch-pushes the // recorded refs fast-forward-only, recovering a diverged ref by fetch+replay and -// removing from the queue only the refs that land. It returns the number pushed. +// removing from the queue only the refs that land. It returns the number pushed +// and the number withheld for want of the OPF trailer. +// +// When requireOPFTrailer is set, a drained ref whose tip does not already carry +// Entire-OPF-Applied is excluded from this push entirely: never attempted, never +// pruned as stale, left queued exactly like a ref whose push was rejected. That +// is how OPF delivery fails closed per ref — a ref still awaiting redaction is +// simply invisible to this delivery attempt — rather than by withholding the +// whole flush and stranding siblings that already carry the trailer. +// +// The check reads the trailer off the refs this call drained, and is made here +// rather than handed in as a set computed earlier for one reason: between any +// earlier readiness read and the Drain below, a concurrent session finalizing a +// checkpoint can enqueue a ref that the earlier read never saw. Such a ref would +// be in this push and in no exclusion list, and would ship un-redacted. Checking +// and using the identical set closes that window instead of narrowing it; see +// deliveryRequiresOPFTrailer. // // Shared by the git-refs pre-push path (which logs and ignores the error to // never block the user's push) and the migration command's opt-in push (which // surfaces it). Stale entries — refs no longer present locally — are pruned so -// they don't block the queue forever. -func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps pushSettings) (int, error) { +// they don't block the queue forever, whether or not the trailer is required: +// an absent ref has nothing to redact and nothing to deliver. +func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps pushSettings, requireOPFTrailer bool) (pushedCount, withheldCount int, err error) { queue, err := checkpoint.PushQueueForRepo(ctx, repo) if err != nil { - return 0, fmt.Errorf("resolve push queue: %w", err) + return 0, 0, fmt.Errorf("resolve push queue: %w", err) } - queued, err := queue.Drain() + queued, err := queue.DrainEntries() if err != nil { - return 0, fmt.Errorf("drain push queue: %w", err) + return 0, 0, fmt.Errorf("drain push queue: %w", err) } if len(queued) == 0 { - return 0, nil + return 0, 0, nil } - pushCtx, pushSpan := perf.Start(ctx, "push_checkpoint_refs") - defer pushSpan.End() - - existing, stale := partitionLocalRefs(repo, queued) + existing, awaiting, stale := partitionCheckpointRefPushes(repo, queued, requireOPFTrailer) if len(stale) > 0 { - if err := queue.Remove(stale); err != nil { + if err := queue.RemoveEntries(stale); err != nil { logging.Warn(ctx, "git-refs push: prune stale queue entries failed", slog.String("error", err.Error())) } } + + // Per-ref delivery gate, on the drained set and nothing else. Skipped + // wholesale when the trailer is not required, so the OPF-off fast path pays + // for no commit loads at all. + withheld := len(awaiting) if len(existing) == 0 { - return 0, nil + return 0, withheld, nil } + pushCtx, pushSpan := perf.Start(ctx, "push_checkpoint_refs") + defer pushSpan.End() + // Resolved here, not by the caller: it spawns `git remote get-url` and its // result is unused unless refs are actually pushed, so an ordinary push with // an empty queue must not pay for it — nor print the multi-URL warning. @@ -525,14 +691,14 @@ func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps push // Fast path: push all refs in one round-trip (fast-forward-only). If every // ref was up to date or fast-forwarded, we're done. - batchErr := batchPushRefs(pushCtx, dest.target, existing) + batchErr := batchPushCheckpointRefs(pushCtx, dest.target, existing) if batchErr == nil { stop(" done") - if removeErr := queue.Remove(existing); removeErr != nil { + if removeErr := queue.RemoveEntries(checkpointRefPushTokens(existing)); removeErr != nil { logging.Warn(ctx, "git-refs push: clear pushed refs from queue failed", slog.String("error", removeErr.Error())) } - return len(existing), nil + return len(existing), withheld, nil } stop("") @@ -549,7 +715,7 @@ func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps push if dest.checkpointRemote { printCheckpointRemoteHint(dest.target) } - return 0, batchErr + return 0, withheld, batchErr } // At least one ref was rejected — typically a non-fast-forward divergence @@ -573,7 +739,7 @@ func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps push fmt.Fprintf(os.Stderr, "[entire] Checkpoint ref push failed; retrying %d ref(s) individually...", len(existing)) stop = startProgressDots(os.Stderr) - pushed := make([]plumbing.ReferenceName, 0, len(existing)) + pushed := make([]checkpointRefPush, 0, len(existing)) var firstErr error var rejectionWarning string // Bounded; see checkpointFlushBudget and maxConsecutiveRefPushFailures. @@ -582,7 +748,7 @@ func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps push consecutiveFailures := 0 attempted := 0 var abortReason string - var failed []plumbing.ReferenceName + var failed []checkpoint.PushQueueEntry for _, ref := range existing { // Before the attempt but never before the first: a flush always tries at // least one ref, and only aborts while refs are left to skip. @@ -595,25 +761,26 @@ func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps push } } attempted++ - if err := pushCheckpointRefWithRecovery(flushCtx, dest.target, ref); err != nil { + landed, pushErr := pushCheckpointRefWithRecovery(flushCtx, repo, dest.target, ref, requireOPFTrailer) + if pushErr != nil { consecutiveFailures++ logging.Warn(ctx, "git-refs push: checkpoint ref push/sync failed; left queued, not overwritten", - slog.String("ref", ref.String()), slog.String("error", err.Error())) - if nonInteractiveSSHAuthFailure(flushCtx, err) { + slog.String("ref", ref.name.String()), slog.String("error", pushErr.Error())) + if nonInteractiveSSHAuthFailure(flushCtx, pushErr) { printNonInteractiveSSHAuthHint() } if rejectionWarning == "" { - if reason := checkpointRefRejectionReason(err); reason != "" { - rejectionWarning = fmt.Sprintf("[entire] Warning: checkpoint ref %s remains queued (showing one rejection):\n%s", ref, reason) + if reason := checkpointRefRejectionReason(pushErr); reason != "" { + rejectionWarning = fmt.Sprintf("[entire] Warning: checkpoint ref %s remains queued (showing one rejection):\n%s", ref.name, reason) } } if firstErr == nil { - firstErr = err + firstErr = pushErr } - failed = append(failed, ref) + failed = append(failed, ref.token) } else { consecutiveFailures = 0 - pushed = append(pushed, ref) + pushed = append(pushed, landed) } } stop(fmt.Sprintf(" pushed %d of %d", len(pushed), len(existing))) @@ -634,7 +801,7 @@ func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps push if rejectionWarning != "" { fmt.Fprintln(os.Stderr, rejectionWarning) } - if err := queue.Remove(pushed); err != nil { + if err := queue.RemoveEntries(checkpointRefPushTokens(pushed)); err != nil { logging.Warn(ctx, "git-refs push: clear pushed refs from queue failed", slog.String("error", err.Error())) } @@ -665,8 +832,8 @@ func flushCheckpointRefsQueue(ctx context.Context, repo *git.Repository, ps push // Counts attempts, not the queue: refs skipped by an early abort were // never tried, and reporting them as failures would overstate what the // remote actually refused. - return len(pushed), fmt.Errorf("%d of %d attempted checkpoint refs failed to push: %w", + return len(pushed), withheld, fmt.Errorf("%d of %d attempted checkpoint refs failed to push: %w", attempted-len(pushed), attempted, firstErr) } - return len(pushed), nil + return len(pushed), withheld, nil } diff --git a/cmd/entire/cli/strategy/manual_commit_push_test.go b/cmd/entire/cli/strategy/manual_commit_push_test.go index 75485f2ed4..53af8b4990 100644 --- a/cmd/entire/cli/strategy/manual_commit_push_test.go +++ b/cmd/entire/cli/strategy/manual_commit_push_test.go @@ -5,8 +5,13 @@ import ( "os/exec" "testing" + "github.com/go-git/go-git/v6/plumbing" + + "github.com/entireio/cli/cmd/entire/cli/checkpoint" + "github.com/entireio/cli/cmd/entire/cli/checkpoint/id" "github.com/entireio/cli/cmd/entire/cli/testutil" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -54,3 +59,157 @@ func TestDeferCheckpointPushOnEmptyRemote_UsesLocalTrackingRefs(t *testing.T) { deferCheckpointPushOnEmptyRemote(ctx, pushSettings{remote: "origin", checkpointURL: "https://example.invalid/cp.git"}), "a dedicated checkpoint remote is exempt from the guard") } + +// TestPrePushCheckpointRefs_RedactedRefShipsWhileFailedSiblingStaysQueued pins +// per-ref delivery on the pre-push path: a ref OPF finished ships in the same +// push even though a sibling could not be redacted, and that sibling stays +// queued without reaching the remote. +func TestPrePushCheckpointRefs_RedactedRefShipsWhileFailedSiblingStaysQueued(t *testing.T) { + // No t.Parallel: the fixture uses t.Chdir. + const okID, failID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + const failSentinel = "OPFBOOM" + configureFakeOPF(t, &fakeRuntimeFailsOnSentinel{sentinel: failSentinel}) + bareDir, repo, refs := setupGitRefsOPFRepo(t, okID) + addGitRefsSessionWithTranscript(t, repo, failID, "sess-fail", + "Hello, PERSONABC asked about "+failSentinel) + refs = append(refs, mustRefName(t, id.MustCheckpointID(failID))) + + require.NoError(t, NewManualCommitStrategy().PrePushFromGitHook(t.Context(), "origin"), + "a ref OPF could not finish must not block the user's git push") + + require.NotEmpty(t, remoteRefHash(t, bareDir, refs[0]), + "the redacted ref must reach the remote") + assertRefsAbsentFromRemote(t, bareDir, []plumbing.ReferenceName{refs[1]}, + "the ref OPF could not finish must not reach the remote") + require.Equal(t, []plumbing.ReferenceName{refs[1]}, queuedRefs(t, repo), + "only the ref OPF could not finish stays queued") +} + +// TestPushQueuedCheckpointRefs_ShipsReadyRefsAndReportsTheRest is the explicit +// push's half of the same gate. It cannot log-and-swallow like the pre-push +// path, so it has to say both things at once: the refs that were ready went, +// AND some are still queued. pushed being non-zero next to a non-nil error is +// the contract `doctor migrate-checkpoints` depends on to report what landed. +func TestPushQueuedCheckpointRefs_ShipsReadyRefsAndReportsTheRest(t *testing.T) { + // No t.Parallel: the fixture uses t.Chdir. + const okID, failID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + const failSentinel = "OPFBOOM" + configureFakeOPF(t, &fakeRuntimeFailsOnSentinel{sentinel: failSentinel}) + bareDir, repo, refs := setupGitRefsOPFRepo(t, okID, failID) + addGitRefsSessionWithTranscript(t, repo, failID, "sess-fail", + "Hello, PERSONABC asked about "+failSentinel) + + pushed, pushDisabled, err := PushQueuedCheckpointRefs(t.Context(), repo, bareDir) + + require.ErrorContains(t, err, "1 checkpoint ref(s) stay queued", + "the leftover must be reported, and counted, not swallowed") + require.False(t, pushDisabled) + require.Equal(t, 1, pushed, "a non-nil error must not hide the ref that actually landed") + + require.NotEmpty(t, remoteRefHash(t, bareDir, refs[0]), + "the already-redacted ref must reach the remote") + require.Equal(t, []plumbing.ReferenceName{refs[1]}, queuedRefs(t, repo), + "only the ref OPF could not finish stays queued") +} + +// TestFlushCheckpointRefsQueue_WithholdsARefEnqueuedAfterTheRewrite walks the +// timeline where a concurrent session enqueues an untrailered ref after the +// inline rewrite finished but before the flush drains the queue. The flush is +// told only WHETHER the trailer is required and reads it off the set it +// drained, so the racing ref is checked like every other and held back rather +// than shipped as 8-layer content under an OPFRun decision. +func TestFlushCheckpointRefsQueue_WithholdsARefEnqueuedAfterTheRewrite(t *testing.T) { + // No t.Parallel: the fixture uses t.Chdir. + const earlyID, lateID = "a1b2c3d4e5f6", "b2c3d4e5f6a1" + configureFakeOPF(t, &fakeOPFForRewrite{}) + bareDir, repo, refs := setupGitRefsOPFRepo(t, earlyID) + earlyRef := refs[0] + + // The gate's inline rewrite: the queued ref is redacted and trailered, so it + // is genuinely ready to ship. + require.NoError(t, RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo)) + + // A concurrent session finalizes a checkpoint, enqueuing a ref that has + // never been near OPF. + addGitRefsSession(t, repo, lateID, "sess-late") + lateRef := mustRefName(t, id.MustCheckpointID(lateID)) + require.ElementsMatch(t, []plumbing.ReferenceName{earlyRef, lateRef}, queuedRefs(t, repo), + "both refs are queued, so both are in what the flush drains") + + // The flush. It is handed no exclusion set, only the requirement. + pushed, withheld, err := flushCheckpointRefsQueue( + t.Context(), repo, pushSettings{remote: bareDir}, true) + require.NoError(t, err) + assert.Equal(t, 1, pushed, "the trailered ref must still ship; per-ref delivery is the point") + assert.Equal(t, 1, withheld, "the racing untrailered ref must be counted as held back") + + assert.NotEmpty(t, remoteRefHash(t, bareDir, earlyRef), + "the already-redacted ref must reach the remote") + assertRefsAbsentFromRemote(t, bareDir, []plumbing.ReferenceName{lateRef}, + "a ref enqueued after the readiness read must not ship unchecked") + assert.Equal(t, []plumbing.ReferenceName{lateRef}, queuedRefs(t, repo), + "the withheld ref stays queued untouched for the next push") +} + +func TestCheckpointRefDelivery_PreservesSameRefGenerationAdvancedAfterVerification(t *testing.T) { + // No t.Parallel: the fixture uses t.Chdir and OPF test configuration. + const checkpointID = "a1b2c3d4e5f6" + configureFakeOPF(t, &fakeOPFForRewrite{}) + bareDir, repo, refs := setupGitRefsOPFRepo(t, checkpointID) + refName := refs[0] + require.NoError(t, RewriteQueuedCheckpointRefsWithOPF(t.Context(), repo)) + + queue, err := checkpoint.PushQueueForRepo(t.Context(), repo) + require.NoError(t, err) + drained, err := queue.DrainEntries() + require.NoError(t, err) + ready, awaiting, stale := partitionCheckpointRefPushes(repo, drained, true) + require.Len(t, ready, 1) + require.Empty(t, awaiting) + require.Empty(t, stale) + verifiedHash := ready[0].hash + + addGitRefsSession(t, repo, checkpointID, "sess-later") + current, err := repo.Reference(refName, true) + require.NoError(t, err) + require.NotEqual(t, verifiedHash, current.Hash()) + + require.NoError(t, batchPushCheckpointRefs(t.Context(), bareDir, ready)) + require.NoError(t, queue.RemoveEntries(checkpointRefPushTokens(ready))) + assert.Equal(t, verifiedHash.String(), remoteRefHash(t, bareDir, refName), + "delivery must stay pinned to the hash that passed the trailer check") + + remaining, err := queue.DrainEntries() + require.NoError(t, err) + require.Equal(t, []checkpoint.PushQueueEntry{{Ref: refName, Hash: current.Hash()}}, remaining, + "cleanup for the delivered generation must preserve the newer generation") + + pushed, withheld, err := flushCheckpointRefsQueue( + t.Context(), repo, pushSettings{remote: bareDir}, true) + require.NoError(t, err) + assert.Zero(t, pushed) + assert.Equal(t, 1, withheld, "the newer untrailered generation must remain fail-closed") + assert.Equal(t, verifiedHash.String(), remoteRefHash(t, bareDir, refName)) +} + +// TestPrePushCheckpointRefs_OPFSkipShipsTheWholeQueueUnchanged is the other half +// of the delivery gate: an explicit opt-out for this push (ENTIRE_OPF=no) is not +// a backlog. Nothing is rewritten, so every queued ref is untrailered — and the +// per-ref filter must not mistake that for "still redacting" and withhold the +// queue the user deliberately chose to ship as-is, 8-layer and untagged. +func TestPrePushCheckpointRefs_OPFSkipShipsTheWholeQueueUnchanged(t *testing.T) { + // No t.Parallel: uses t.Setenv and the fixture's t.Chdir. + configureFakeOPF(t, &fakeOPFForRewrite{}) + t.Setenv("ENTIRE_OPF", "no") + bareDir, repo, refs := setupGitRefsOPFRepo(t, "a1b2c3d4e5f6", "b2c3d4e5f6a1") + before := refHashes(t, repo, refs) + + require.NoError(t, NewManualCommitStrategy().PrePushFromGitHook(t.Context(), "origin")) + + require.Equal(t, before, refHashes(t, repo, refs), "an opted-out push must rewrite nothing") + for i, ref := range refs { + require.Equal(t, before[i].String(), remoteRefHash(t, bareDir, ref), + "every ref must ship as written when the user opted out of OPF") + } + require.Empty(t, queuedRefs(t, repo), "pushed refs leave the queue") +} diff --git a/cmd/entire/cli/strategy/push_common.go b/cmd/entire/cli/strategy/push_common.go index 79b832fb0a..fe6cc0ae1e 100644 --- a/cmd/entire/cli/strategy/push_common.go +++ b/cmd/entire/cli/strategy/push_common.go @@ -12,9 +12,11 @@ import ( "sync" "time" + "github.com/entireio/cli/cmd/entire/cli/checkpoint" "github.com/entireio/cli/cmd/entire/cli/checkpoint/remote" "github.com/entireio/cli/cmd/entire/cli/logging" "github.com/entireio/cli/cmd/entire/cli/settings" + "github.com/entireio/cli/cmd/entire/cli/trailers" "github.com/entireio/cli/perf" "github.com/go-git/go-git/v6" @@ -22,6 +24,20 @@ import ( "github.com/go-git/go-git/v6/plumbing/object" ) +// checkpointRefPush binds delivery to one immutable source hash while retaining +// the queue generation token that successful delivery may remove. +type checkpointRefPush struct { + token checkpoint.PushQueueEntry + name plumbing.ReferenceName + hash plumbing.Hash +} + +var checkpointRefRecoveryCAS = checkpoint.CASPersistentRef //nolint:gochecknoglobals // deterministic recovery-race test seam + +func (p checkpointRefPush) refSpec() string { + return p.hash.String() + ":" + p.name.String() +} + // partitionLocalRefs splits refs into those that exist locally (pushable) and // those that don't (stale queue entries — e.g. a checkpoint ref deleted by // cleanup). Stale refs can never push, so callers drop them from the queue @@ -69,6 +85,20 @@ func batchPushRefs(ctx context.Context, target string, refs []plumbing.Reference return nil } +func batchPushCheckpointRefs(ctx context.Context, target string, refs []checkpointRefPush) error { + if len(refs) == 0 { + return nil + } + refSpecs := make([]string, 0, len(refs)) + for _, ref := range refs { + refSpecs = append(refSpecs, ref.refSpec()) + } + if _, err := remote.PushWithOptions(ctx, remote.PushOptions{Remote: target, RefSpecs: refSpecs}); err != nil { + return fmt.Errorf("push %d checkpoint refs: %w", len(refs), err) + } + return nil +} + // pushCheckpointRefWithRecovery pushes a single checkpoint ref fast-forward-only; // confirmed remote policy/hook rejections return immediately. Other failures — // typically the ref diverged on the remote (the same checkpoint re-written @@ -80,28 +110,172 @@ func batchPushRefs(ctx context.Context, target string, refs []plumbing.Reference // (e.g. both sides rewrote the root metadata.json) surfaces as a rebase error and // the ref is left for a later pre-push. Returns nil only if the ref reached the // remote. -func pushCheckpointRefWithRecovery(ctx context.Context, target string, ref plumbing.ReferenceName) error { +func pushCheckpointRefWithRecovery( + ctx context.Context, + repo *git.Repository, + target string, + candidate checkpointRefPush, + requireOPFTrailer bool, +) (checkpointRefPush, error) { // One shared budget across the initial push, fetch+replay, and retry, matching // doPushRef (fetchAndRebaseRefCommon relies on the caller's deadline). ctx, cancel := context.WithTimeout(ctx, checkpointPushBudget) defer cancel() - pushErr := batchPushRefs(ctx, target, []plumbing.ReferenceName{ref}) + pushErr := batchPushCheckpointRefs(ctx, target, []checkpointRefPush{candidate}) if pushErr == nil { - return nil + return candidate, nil } if checkpointRefRejectionReason(pushErr) != "" { // Fetch+replay cannot fix a remote policy/hook rejection. If the ref // already exists remotely, replay would needlessly rewrite local // commits (including their committer timestamps) on every push. - return pushErr + return checkpointRefPush{}, pushErr } - if err := fetchAndRebaseRefCommon(ctx, target, ref); err != nil { + recovered, onRemote, err := recoverCheckpointRef(ctx, repo, target, candidate) + if err != nil { // Recovery is speculative: a missing remote ref may mean the push was // blocked, not that it diverged. Keep the push failure primary. - return &checkpointRefRecoveryError{pushErr: pushErr, recoveryErr: err} + return checkpointRefPush{}, &checkpointRefRecoveryError{pushErr: pushErr, recoveryErr: err} + } + if onRemote { + // The remote already holds everything the candidate carried, so nothing + // of ours is left to ship. The tip is now the remote's own commit, which + // the trailer check must not judge: its history was published elsewhere. + return recovered, nil + } + if requireOPFTrailer { + commit, commitErr := repo.CommitObject(recovered.hash) + if commitErr != nil { + return checkpointRefPush{}, fmt.Errorf("load recovered checkpoint ref %s: %w", candidate.name, commitErr) + } + if !trailers.HasOPFApplied(commit.Message) { + return checkpointRefPush{}, fmt.Errorf("recovered checkpoint ref %s does not carry the OPF trailer", candidate.name) + } + } + if err := batchPushCheckpointRefs(ctx, target, []checkpointRefPush{recovered}); err != nil { + return checkpointRefPush{}, err + } + return recovered, nil +} + +// recoverCheckpointRef replays exactly candidate.hash onto the fetched remote +// tip and installs the result only if the local ref still names that same +// generation. A concurrent checkpoint write therefore wins the CAS and remains +// both locally reachable and queued for a later delivery attempt. onRemote +// reports that the replay added nothing: the remote tip already contains the +// candidate, and the local ref now names that tip. +func recoverCheckpointRef( + ctx context.Context, + repo *git.Repository, + target string, + candidate checkpointRefPush, +) (recovered checkpointRefPush, onRemote bool, err error) { + fetchTarget, err := remote.ResolveFetchTarget(ctx, target) + if err != nil { + return checkpointRefPush{}, false, fmt.Errorf("resolve fetch target: %w", err) + } + fetchedRefName := plumbing.ReferenceName( + "refs/entire-fetch-tmp/" + strings.TrimPrefix(candidate.name.String(), "refs/"), + ) + refSpec := fmt.Sprintf("+%s:%s", candidate.name, fetchedRefName) + _, fetchSpan := perf.Start(ctx, "git_fetch") + _, fetchErr := remote.Fetch(ctx, remote.FetchOptions{ + Remote: fetchTarget, + RefSpecs: []string{refSpec}, + NoTags: true, + }) + fetchSpan.RecordError(fetchErr) + fetchSpan.End() + if fetchErr != nil { + return checkpointRefPush{}, false, fmt.Errorf("fetch failed: %w", fetchErr) + } + defer func() { + _ = repo.Storer.RemoveReference(fetchedRefName) //nolint:errcheck // cleanup is best-effort + }() + + remoteRef, err := repo.Reference(fetchedRefName, true) + if err != nil { + return checkpointRefPush{}, false, fmt.Errorf("resolve fetched checkpoint ref: %w", err) + } + newTip, err := replayCheckpointCandidate(ctx, repo, candidate.hash, remoteRef.Hash()) + if err != nil { + return checkpointRefPush{}, false, err + } + if err := checkpointRefRecoveryCAS(ctx, repo, candidate.name, newTip, candidate.hash); err != nil { + // newTip has no unique writer data: it is derived entirely from the + // immutable candidate plus the fetched remote. On a CAS conflict the + // writer's winning generation remains the live ref and its queue entry + // remains untouched. A later delivery recomputes the replay from that + // winner; retaining this losing intermediate under another ref would + // create a second source of truth for the checkpoint. + return checkpointRefPush{}, false, fmt.Errorf("install recovered checkpoint ref %s: %w", candidate.name, err) + } + candidate.hash = newTip + return candidate, newTip.Equal(remoteRef.Hash()), nil +} + +func replayCheckpointCandidate( + ctx context.Context, + repo *git.Repository, + localHash, remoteHash plumbing.Hash, +) (plumbing.Hash, error) { + if localHash.Equal(remoteHash) { + return remoteHash, nil + } + repoPath, err := getRepoPath(repo) + if err != nil { + return plumbing.ZeroHash, fmt.Errorf("get repository path: %w", err) + } + mergeBase, err := getMergeBase(ctx, repoPath, localHash.String(), remoteHash.String()) + if err == nil && mergeBase.Equal(localHash) { + return remoteHash, nil + } + + shallow, shallowErr := loadShallowHashes(ctx, repoPath) + if shallowErr != nil { + return plumbing.ZeroHash, fmt.Errorf("load shallow boundaries: %w", shallowErr) + } + if errors.Is(err, errNoMergeBase) { + commits, collectErr := collectCommitChain(repo, localHash, shallow) + if collectErr != nil { + return plumbing.ZeroHash, fmt.Errorf("collect disconnected checkpoint commits: %w", collectErr) + } + dataCommits := make([]*object.Commit, 0, len(commits)) + for _, commit := range commits { + tree, treeErr := commit.Tree() + if treeErr != nil { + return plumbing.ZeroHash, fmt.Errorf("read tree for commit %s: %w", commit.Hash.String()[:7], treeErr) + } + if len(tree.Entries) > 0 { + dataCommits = append(dataCommits, commit) + } + } + if len(dataCommits) == 0 { + return remoteHash, nil + } + newTip, pickErr := cherryPickOnto(ctx, repo, remoteHash, dataCommits, shallow) + if pickErr != nil { + return plumbing.ZeroHash, fmt.Errorf("replay disconnected checkpoint commits: %w", pickErr) + } + return newTip, nil + } + if err != nil { + return plumbing.ZeroHash, fmt.Errorf("find checkpoint merge base: %w", err) + } + + commits, err := collectCommitsSince(ctx, repo, repoPath, localHash, remoteHash) + if err != nil { + return plumbing.ZeroHash, fmt.Errorf("collect local checkpoint commits: %w", err) + } + if len(commits) == 0 { + return remoteHash, nil + } + newTip, err := cherryPickOnto(ctx, repo, remoteHash, commits, shallow) + if err != nil { + return plumbing.ZeroHash, fmt.Errorf("replay checkpoint commits: %w", err) } - return batchPushRefs(ctx, target, []plumbing.ReferenceName{ref}) + return newTip, nil } // checkpointRefRecoveryError keeps both failures inspectable, while allowing the diff --git a/cmd/entire/cli/strategy/refs_push_rejection_test.go b/cmd/entire/cli/strategy/refs_push_rejection_test.go index cc18a0fdbf..a0121cb1bf 100644 --- a/cmd/entire/cli/strategy/refs_push_rejection_test.go +++ b/cmd/entire/cli/strategy/refs_push_rejection_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" + "github.com/entireio/cli/cmd/entire/cli/checkpoint" "github.com/entireio/cli/cmd/entire/cli/gitrepo" "github.com/entireio/cli/cmd/entire/cli/paths" "github.com/entireio/cli/cmd/entire/cli/testutil" @@ -42,7 +43,17 @@ func TestPushCheckpointRefWithRecovery_PreservesRejection(t *testing.T) { t.Chdir(workDir) // CWD-based push/recovery; cannot run in parallel. installCheckpointRejectHook(t, bareDir, false, "") - err := pushCheckpointRefWithRecovery(t.Context(), bareDir, refs[0]) + repo, err := gitrepo.OpenPath(workDir) + require.NoError(t, err) + defer repo.Close() + ref, err := repo.Reference(refs[0], true) + require.NoError(t, err) + candidate := checkpointRefPush{ + token: checkpoint.PushQueueEntry{Ref: refs[0], Hash: ref.Hash()}, + name: refs[0], + hash: ref.Hash(), + } + _, err = pushCheckpointRefWithRecovery(t.Context(), repo, bareDir, candidate, false) require.ErrorContains(t, err, checkpointRejectReason) require.ErrorContains(t, err, checkpointUnblockURL) assert.NotContains(t, err.Error(), "sync diverged") @@ -57,7 +68,17 @@ func TestPushCheckpointRefWithRecovery_PreservesUnknownFailure(t *testing.T) { t.Chdir(workDir) // Unknown failures still attempt recovery and must preserve both causes. - err := pushCheckpointRefWithRecovery(t.Context(), filepath.Join(t.TempDir(), "missing.git"), refs[0]) + repo, err := gitrepo.OpenPath(workDir) + require.NoError(t, err) + defer repo.Close() + ref, err := repo.Reference(refs[0], true) + require.NoError(t, err) + candidate := checkpointRefPush{ + token: checkpoint.PushQueueEntry{Ref: refs[0], Hash: ref.Hash()}, + name: refs[0], + hash: ref.Hash(), + } + _, err = pushCheckpointRefWithRecovery(t.Context(), repo, filepath.Join(t.TempDir(), "missing.git"), candidate, false) var recoveryErr *checkpointRefRecoveryError require.ErrorAs(t, err, &recoveryErr) require.ErrorIs(t, err, recoveryErr.pushErr) diff --git a/cmd/entire/cli/strategy/refs_push_test.go b/cmd/entire/cli/strategy/refs_push_test.go index dc03bb7806..7233b29a4a 100644 --- a/cmd/entire/cli/strategy/refs_push_test.go +++ b/cmd/entire/cli/strategy/refs_push_test.go @@ -65,6 +65,36 @@ func TestBatchPushRefs(t *testing.T) { } } +func TestBatchPushCheckpointRefs_PinsVerifiedHash(t *testing.T) { + workDir, bareDir, refs := setupRepoWithCheckpointRefs(t) + t.Chdir(workDir) + repo, err := git.PlainOpen(workDir) + require.NoError(t, err) + refName := refs[0] + verified, err := repo.Reference(refName, true) + require.NoError(t, err) + + testutil.WriteFile(t, workDir, "later.txt", "later") + testutil.GitAdd(t, workDir, "later.txt") + testutil.GitCommit(t, workDir, "later") + later, err := repo.Head() + require.NoError(t, err) + require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(refName, later.Hash()))) + + candidate := checkpointRefPush{ + token: checkpoint.PushQueueEntry{Ref: refName, Hash: verified.Hash()}, + name: refName, + hash: verified.Hash(), + } + require.NoError(t, batchPushCheckpointRefs(t.Context(), bareDir, []checkpointRefPush{candidate})) + + assert.Equal(t, verified.Hash().String(), remoteRefHash(t, bareDir, refName), + "the remote must receive the captured hash, not the ref's later tip") + current, err := repo.Reference(refName, true) + require.NoError(t, err) + assert.Equal(t, later.Hash(), current.Hash(), "pushing the captured hash must not rewind the local ref") +} + func TestBatchPushRefs_Empty(t *testing.T) { t.Parallel() // No refs → no git invocation, no error. @@ -166,10 +196,11 @@ func TestPushCheckpointRefWithRecovery_MergesDivergedRef(t *testing.T) { // pushes are rejected, then recovery replays C3's delta onto C2. queue := enqueueRefs(t, repo, []plumbing.ReferenceName{ref}) restore := captureStderr(t) - pushed, pushErr := flushCheckpointRefsQueue(ctx, repo, pushSettings{remote: bareDir}) + pushed, withheld, pushErr := flushCheckpointRefsQueue(ctx, repo, pushSettings{remote: bareDir}, false) output := restore() require.NoError(t, pushErr, "diverged ref should be recovered by fetch+replay, not rejected") assert.Equal(t, 1, pushed) + assert.Zero(t, withheld, "no OPF trailer is required here, so nothing is withheld") assert.NotContains(t, output, "Warning:", "plain divergence should recover quietly") remaining, err := queue.Drain() require.NoError(t, err) @@ -180,6 +211,137 @@ func TestPushCheckpointRefWithRecovery_MergesDivergedRef(t *testing.T) { assert.Contains(t, files, "c.txt", "local-only change must be replayed on top") } +// A stale local ref whose commits the remote already contains is delivered, +// even when another clone (OPF off) advanced the remote with untrailered +// commits: the trailer check covers only what this push would ship, and after +// recovery nothing is left to ship. +func TestPushCheckpointRefWithRecovery_RemoteAlreadyContainsLocalIsDelivered(t *testing.T) { + workDir, bareDir, refs := setupRepoWithCheckpointRefs(t) + t.Chdir(workDir) + ctx := context.Background() + ref := refs[0] + + repo, err := git.PlainOpen(workDir) + require.NoError(t, err) + runGit := func(args ...string) string { + return strings.TrimSpace(testutil.RunGit(t, workDir, args...)) + } + tree := runGit("rev-parse", "HEAD^{tree}") + local := plumbing.NewHash(runGit("commit-tree", tree, "-p", "HEAD", + "-m", trailers.AppendOPFAppliedTrailer("local checkpoint"))) + remoteTip := plumbing.NewHash(runGit("commit-tree", tree, "-p", local.String(), "-m", "written with OPF off")) + require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(ref, remoteTip))) + require.NoError(t, batchPushRefs(ctx, bareDir, []plumbing.ReferenceName{ref})) + require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(ref, local))) + queue := enqueueRefs(t, repo, []plumbing.ReferenceName{ref}) + + restore := captureStderr(t) + pushed, withheld, pushErr := flushCheckpointRefsQueue(ctx, repo, pushSettings{remote: bareDir}, true) + restore() + + require.NoError(t, pushErr, "the remote already holds the local commit; nothing is left to verify") + assert.Equal(t, 1, pushed) + assert.Zero(t, withheld) + remaining, err := queue.Drain() + require.NoError(t, err) + assert.Empty(t, remaining, "the delivered ref must leave the queue") + assert.Equal(t, remoteTip.String(), remoteRefHash(t, bareDir, ref), "the remote must not move") +} + +func TestPushCheckpointRefWithRecovery_PreservesConcurrentGenerationOnCASConflict(t *testing.T) { + workDir, bareDir, refs := setupRepoWithCheckpointRefs(t) + t.Chdir(workDir) + ctx := context.Background() + ref := refs[0] + + repo, err := git.PlainOpen(workDir) + require.NoError(t, err) + head, err := repo.Head() + require.NoError(t, err) + base := head.Hash() + require.NoError(t, batchPushRefs(ctx, bareDir, []plumbing.ReferenceName{ref})) + + // Advance the remote from base. + testutil.WriteFile(t, workDir, "remote.txt", "remote") + testutil.GitAdd(t, workDir, "remote.txt") + testutil.GitCommit(t, workDir, "remote generation") + head, err = repo.Head() + require.NoError(t, err) + require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(ref, head.Hash()))) + require.NoError(t, batchPushRefs(ctx, bareDir, []plumbing.ReferenceName{ref})) + remoteTip := head.Hash() + + // Create a disconnected queued generation that recovery will try to replay. + // This covers the former ReconcileDisconnectedMetadataRef path, whose final + // update was unconditional before checkpoint recovery became generation-safe. + testutil.GitReset(t, workDir, base.String()) + testutil.WriteFile(t, workDir, "local.txt", "local") + testutil.GitAdd(t, workDir, "local.txt") + tree := strings.TrimSpace(testutil.RunGit(t, workDir, "write-tree")) + candidate := plumbing.NewHash(strings.TrimSpace(testutil.RunGit(t, workDir, "commit-tree", tree, "-m", "local generation"))) + require.NoError(t, repo.Storer.SetReference(plumbing.NewHashReference(ref, candidate))) + queue := enqueueRefs(t, repo, []plumbing.ReferenceName{ref}) + + // At the recovery install boundary, simulate a checkpoint writer advancing + // the same ref and queueing that newer generation. The real CAS must reject + // the stale recovery rather than overwrite it. + oldCAS := checkpointRefRecoveryCAS + var concurrent plumbing.Hash + checkpointRefRecoveryCAS = func( + casCtx context.Context, + casRepo *git.Repository, + refName plumbing.ReferenceName, + newHash, expectedOld plumbing.Hash, + ) error { + require.Equal(t, expectedOld, candidate) + runGit := func(args ...string) string { + return strings.TrimSpace(testutil.RunGit(t, workDir, args...)) + } + testutil.WriteFile(t, workDir, "concurrent.txt", "concurrent") + testutil.GitAdd(t, workDir, "concurrent.txt") + concurrentTree := runGit("write-tree") + concurrent = plumbing.NewHash(runGit("commit-tree", concurrentTree, "-p", candidate.String(), "-m", "concurrent generation")) + require.NoError(t, casRepo.Storer.SetReference(plumbing.NewHashReference(refName, concurrent))) + require.NoError(t, queue.EnqueueEntry(checkpoint.PushQueueEntry{Ref: refName, Hash: concurrent})) + return checkpoint.CASPersistentRef(casCtx, casRepo, refName, newHash, expectedOld) + } + t.Cleanup(func() { checkpointRefRecoveryCAS = oldCAS }) + + restore := captureStderr(t) + pushed, withheld, pushErr := flushCheckpointRefsQueue(ctx, repo, pushSettings{remote: bareDir}, false) + restore() + require.Error(t, pushErr) + assert.Zero(t, pushed) + assert.Zero(t, withheld) + + localRef, err := repo.Reference(ref, true) + require.NoError(t, err) + assert.Equal(t, concurrent, localRef.Hash(), "recovery must not overwrite the concurrent checkpoint generation") + assert.Equal(t, remoteTip.String(), remoteRefHash(t, bareDir, ref), "a CAS conflict must not push a stale recovered tip") + queued, err := queue.PeekEntries() + require.NoError(t, err) + require.Len(t, queued, 1) + assert.Equal(t, concurrent, queued[0].Hash, "the concurrent generation must remain queued") + + // The losing recovered commit is only a derived intermediate: the next + // delivery replays the queued winner onto the remote and preserves all three + // sources of data without keeping that stale intermediate under another ref. + checkpointRefRecoveryCAS = oldCAS + restore = captureStderr(t) + pushed, withheld, pushErr = flushCheckpointRefsQueue(ctx, repo, pushSettings{remote: bareDir}, false) + restore() + require.NoError(t, pushErr) + assert.Equal(t, 1, pushed) + assert.Zero(t, withheld) + queued, err = queue.PeekEntries() + require.NoError(t, err) + assert.Empty(t, queued) + files := remoteRefFiles(t, bareDir, ref) + assert.Contains(t, files, "remote.txt") + assert.Contains(t, files, "local.txt") + assert.Contains(t, files, "concurrent.txt") +} + // enqueueRefs seeds the repo's push queue with the given refs. func enqueueRefs(t *testing.T, repo *git.Repository, refs []plumbing.ReferenceName) *checkpoint.PushQueue { t.Helper() diff --git a/docs/security-and-privacy.md b/docs/security-and-privacy.md index d34a0cdfae..e90cbb65bf 100644 --- a/docs/security-and-privacy.md +++ b/docs/security-and-privacy.md @@ -240,7 +240,7 @@ OPF failures at push time are **fail-closed**: if OPF is not on PATH, fails to s How that is enforced depends on the checkpoint backend, because they have different escape hatches: - **git-branch**: the rewrite aborts the push with `OPF runtime failed during pre-push rewrite (command=…); aborting push so regex-only content isn't tagged as OPF-applied`. Your `git push` exits non-zero. The checkpoint branch travels with that push, so refusing the push is the only way to withhold it. -- **git-refs**: checkpoint refs are pushed separately from your branch and stay queued when they are not flushed, so the failure withholds the checkpoint push and lets your own `git push` succeed. Nothing under-redacted ships either way. This is not silent: a warning names the failure and states that checkpoint refs stayed queued for the next push. +- **git-refs**: checkpoint refs are pushed separately from your branch and stay queued when they are not flushed, so the failure withholds the checkpoint refs OPF did not finish and lets your own `git push` succeed. Refs already rewritten earlier in the same push still ship. Nothing under-redacted ships either way. This is not silent: a warning names the failure and states that checkpoint refs stayed queued for the next push. (The circuit breaker is per-process, so a broken install costs one warning instead of one timeout per blob — but the push still aborts.) @@ -279,7 +279,7 @@ The rewrite refuses to proceed in a divergent or oversized state rather than sil **Divergence.** On `git-branch`, if local `entire/checkpoints/v1` has commits that aren't ancestors of the remote's v1, the hook exits with a `entire/checkpoints/v1 has diverged from remote` error. Fetch the remote and either reset local v1 to the remote tip or resolve manually before pushing. -`git-refs` has no divergence pre-check, and doesn't need one: each checkpoint is its own ref, so divergence shows up at push time as a non-fast-forward rejection for that one ref. Recovery fetches the remote ref and replays the local-only commits on top, then retries — still without forcing, so the remote commit is preserved as an ancestor. A ref that can't be replayed stays queued. +`git-refs` has no divergence pre-check, and doesn't need one: each checkpoint is its own ref, so divergence shows up at push time as a non-fast-forward rejection for that one ref. Recovery fetches the remote ref, replays the exact candidate's local-only commits on top, and installs the result only if the local ref still names that candidate, so a concurrent checkpoint write stays reachable and queued rather than being overwritten. It then retries with the recovered hash — still without forcing, so the remote commit is preserved as an ancestor. A ref that can't be replayed or installed stays queued. **Un-OPF'd commit cap: `100` by default.** Override per push: @@ -291,7 +291,7 @@ set -x ENTIRE_OPF_BOOTSTRAP_LIMIT unlimited; git push The two backends trip this differently. On `git-branch` it applies **only on bootstrap** — the first push, when the remote has no v1 yet — counted across all unpushed commits. On `git-refs` it applies on **every** push, counted **per queued ref** over that ref's un-trailered ancestry. In practice the `git-refs` trigger is "OPF was enabled late" or "checkpoints were just migrated from the branch", not "first push". -**Batch cap: `2 MiB` of cumulative prose-leaf content by default** (≈110s of inference), on both backends. An `OPF would run inference on …` error means you've hit it: +**Batch cap: `2 MiB` of prose-leaf content by default** (≈110s of inference), counted per checkpoint ref on `git-refs` and across the whole unpushed chain on `git-branch`. Identical leaves are counted once, matching what OPF actually processes. An `OPF would run inference on …` error means you've hit it: ```fish set -x ENTIRE_OPF_BATCH_LIMIT 10485760; git push # 10 MiB @@ -299,11 +299,11 @@ set -x ENTIRE_OPF_BATCH_LIMIT 10485760; git push # 10 MiB set -x ENTIRE_OPF_BATCH_LIMIT unlimited; git push ``` -**Raw-byte cap: `200 MiB` of blob content buffered in memory**, on both backends. It has no env var of its own — it is derived as 100× the batch cap, so raising `ENTIRE_OPF_BATCH_LIMIT` raises it too. It is checked incrementally as blobs load, so on a pathological push (one commit carrying a huge pasted transcript) this is the cap that fires first. +**Raw-byte cap: `200 MiB` of blob content buffered in memory**, counted per checkpoint ref on `git-refs` (one ref is collected, scanned, rebuilt, and released before the next is loaded) and across the whole unpushed chain on `git-branch`. It has no env var of its own — it is derived as 100× the batch cap, so raising `ENTIRE_OPF_BATCH_LIMIT` raises it too. Each blob's size is checked against the remaining budget before its content is read, so on a pathological push (one commit carrying a huge pasted transcript) this is the cap that fires first, before the oversized allocation. -The three caps protect different failure modes: the commit cap stops "100 throwaway commits", the batch cap stops "one commit with 50 MB of prose", and the raw-byte cap stops the loader exhausting memory before either of the others can be evaluated. On `git-refs` the two byte caps are cumulative across the whole flush (all queued refs together) while the commit cap is per ref. +The three caps protect different failure modes: the commit cap stops "100 throwaway commits", the batch cap stops "one commit with 50 MB of prose", and the raw-byte cap stops the loader exhausting memory before either of the others can be evaluated. On `git-refs` all three caps are per ref: a ref that trips one stays queued untouched while its siblings are still rewritten and ship. -**Concurrent push** from another worktree: both backends compare-and-swap the local ref. If another process moved it while OPF was running, `git-branch` exits with `entire/checkpoints/v1 moved during OPF rewrite …; re-run 'git push' (no fetch needed; the move was local)` and aborts. On `git-refs` the affected ref simply stays queued and the next push picks it up. Note that the `git-refs` rewrite rebuilds every commit before touching any ref, but the ref updates themselves are not atomic *across* refs: a conflict partway through leaves the earlier refs already rewritten. They stay queued and push OPF-applied next time, so this is safe — just not "nothing moved". +**Concurrent push** from another worktree: both backends compare-and-swap the local ref. If another process moved it while OPF was running, `git-branch` exits with `entire/checkpoints/v1 moved during OPF rewrite …; re-run 'git push' (no fetch needed; the move was local)` and aborts. On `git-refs` the affected generation stays queued and the next push picks it up. Each ref is collected, scanned, rebuilt, and compare-and-swapped before the next one starts, so updates are not atomic *across* refs: earlier refs can already be rewritten when a later one conflicts. Delivery pushes the exact hash whose trailer it checked and removes only that generation's queue entry, so a concurrent advance of the same ref can neither substitute unchecked content nor be erased from the queue. #### Persistence of un-redacted-by-OPF content @@ -330,7 +330,7 @@ Two notes on the `git-refs` rows: Two `git-branch`-shaped leftovers are easy to miss once a repo has moved on: - **A `git-branch` mirror never gets OPF'd.** Mirrors receive best-effort write fan-out only and never ref-level mutations, so a mirror branch keeps 8-layer content indefinitely. It isn't pushed at pre-push, so it doesn't reach the remote — but it is local content, and it has a `refs/heads/` reflog. -- **Migration doesn't delete the old branch.** `entire doctor migrate-checkpoints` imports from `entire/checkpoints/v1` and leaves it in place, 8-layer, along with its reflog. Migrated refs also carry no OPF trailer, so the first push after a migration re-OPFs every migrated ref — one commit per ref keeps the commit cap happy, but the 2 MiB batch cap is the realistic trip point. +- **Migration doesn't delete the old branch.** `entire doctor migrate-checkpoints` imports from `entire/checkpoints/v1` and leaves it in place, 8-layer, along with its reflog. Migrated refs also carry no OPF trailer, so the first push after a migration re-OPFs every migrated ref — one commit per ref keeps the commit cap happy, and the 2 MiB batch cap is counted per ref, so a large migrated session is withheld on its own rather than blocking the others. `.entire/metadata//full.jsonl` is Entire's own local working copy of the transcript, written mode `0600`. It is *sanitized* (agent state that cannot be replayed out of a checkpoint is stripped) but **not redacted** — redaction happens on the way into a git object, not on this file. It is the input condensation reads from when the live agent transcript is no longer available. diff --git a/redact/batch.go b/redact/batch.go index 26f362ade8..0455370bd1 100644 --- a/redact/batch.go +++ b/redact/batch.go @@ -1,7 +1,6 @@ package redact import ( - "bytes" "context" "fmt" "strings" @@ -182,41 +181,37 @@ func isJSONLikeName(name string) bool { return strings.HasSuffix(name, ".jsonl") || strings.HasSuffix(name, ".json") } -// SumProseLeafBytes returns the cumulative byte size of prose-shaped -// (has-space) leaves across inputs — the upper bound on what -// BatchBytesWithPrivacyFilter would send to OPF inference. +// SumProseLeafBytes returns the deduplicated byte size of unique +// prose-shaped (has-space) leaves across inputs — exactly what +// BatchBytesWithPrivacyFilter will send to OPF inference. // // Callers use this to enforce a cap before paying the OPF cost: a // push with 100MB of mostly-structural JSON has tens of KB of actual // leaves; a push with 100MB of dense prose has hundreds of MB. The // blob-byte size doesn't tell you which without looking inside. // -// Returns a CONSERVATIVE UPPER BOUND on what would go to OPF — same -// has-space gate and JSONL/JSON parse with whole-content fallback as -// the collector inside BatchBytesWithPrivacyFilter, BUT this function -// does NOT deduplicate identical leaves across blobs. The actual batch -// sent to OPF dedups by leaf-text, so a push with many repeated leaves -// will report higher byte counts here than OPF actually sees. Callers -// using this for cap enforcement get an over-strict bound, which is -// safe (false positives possible, false negatives impossible). +// Uses the same has-space gate, JSONL/JSON leaf extraction, and +// leaf-text dedup as Pass 1 of BatchBytesWithPrivacyFilter, so this is +// not a conservative bound — it is the actual count OPF will process. +// A checkpoint ref's ancestry can repeat the same blob (e.g. an +// unchanged full.jsonl carried across several commits); counting it +// once here matches the single inference result the batch call reuses +// for every repeat. func SumProseLeafBytes(inputs []NamedBlob) int { + seen := make(map[string]struct{}) var total int - for _, in := range inputs { - if isJSONLikeName(in.Name) { - if _, err := jsonlContentImpl(string(in.Content), func(v string) string { - if strings.ContainsRune(v, ' ') { - total += len(v) - } - return v - }, concurrencyUnsafeRedactor); err == nil { - continue - } - // JSON parse failed — fall through to whole-content (matches - // the collector's fallback in BatchBytesWithPrivacyFilter). + addLeaf := func(v string) { + if !strings.ContainsRune(v, ' ') { + return } - if bytes.ContainsRune(in.Content, ' ') { - total += len(in.Content) + if _, ok := seen[v]; ok { + return } + seen[v] = struct{}{} + total += len(v) + } + for _, in := range inputs { + collectLeaves(in, addLeaf) } return total } diff --git a/redact/batch_test.go b/redact/batch_test.go index 0e37434c78..8ad1cb1e5a 100644 --- a/redact/batch_test.go +++ b/redact/batch_test.go @@ -460,6 +460,29 @@ func TestSumProseLeafBytes_CountsAcrossBlobShapes(t *testing.T) { } } +// TestSumProseLeafBytes_DedupsRepeatedLeafAcrossBlobs pins the cap +// measurement to what BatchBytesWithPrivacyFilter actually sends to +// OPF. A checkpoint ref's ancestry commonly repeats the same +// full.jsonl blob across several commits; the batch call dedups by +// leaf text, so counting the repeat again here would charge the cap +// for inference the runtime never performs. +func TestSumProseLeafBytes_DedupsRepeatedLeafAcrossBlobs(t *testing.T) { + t.Parallel() + const leaf = "the quick brown fox jumps" + repeated := `{"role":"user","content":"` + leaf + `"}` + "\n" + inputs := []NamedBlob{ + {Name: "0/full.jsonl", Content: []byte(repeated)}, + {Name: "1/full.jsonl", Content: []byte(repeated)}, // byte-identical blob, different path + } + + got := SumProseLeafBytes(inputs) + if want := len(leaf); got != want { + t.Errorf("SumProseLeafBytes with a leaf repeated across blobs = %d, want %d "+ + "(identical leaf text must be counted once, matching what "+ + "BatchBytesWithPrivacyFilter sends to OPF)", got, want) + } +} + // TestBatchBytesWithPrivacyFilter_NoEnabledCategoriesFailsClosed // covers the misconfiguration where OPF is enabled but the effective // category set is empty (empty map, all-false, or omitted). The model