From ace401b04c743625bcc8a18a2dbf57a6771f36f3 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Wed, 28 Oct 2026 18:27:25 +0000 Subject: [PATCH] fix(git): keep a withheld apply's values out of the next edit S1 withholds an applier's apply that lands over an edit decided after the flush mark, and a Repair puts the edit back. But the apply did land, so the next event of that object carried the apply's value for every field its author did not change, and the window committed the whole object under that author's name: the overwritten edit was lost, and the Repair applied the loss. The worker now keeps each object's last accepted version (the last event an apply window or a flush recorded as an edit) and, on a withhold, records each field where the withheld object differs from it: the accepted value and the applier's, either possibly absent. Every later event of the object, and every resync snapshot of it, is restored against that record: a field still holding the applier's value is set back to the accepted one; a field holding any other value leaves the record; a delete clears it. Co-Authored-By: Claude Opus 5.5 --- docs/TODO.md | 11 +- docs/design/applier-echoes.md | 21 +- internal/git/applier_claim.go | 5 +- internal/git/apply_window.go | 14 +- internal/git/apply_window_carry_test.go | 287 +++++++++++++++++----- internal/git/branch_worker.go | 20 +- internal/git/resync_flush.go | 3 + internal/git/withheld_fields.go | 314 ++++++++++++++++++++++++ 8 files changed, 591 insertions(+), 84 deletions(-) create mode 100644 internal/git/withheld_fields.go diff --git a/docs/TODO.md b/docs/TODO.md index 87820ea6..893a395b 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -7,16 +7,13 @@ should land before the next tag is cut (the last release is v0.55.0). Checked ag ## Before the next release -- [ ] **A withheld apply rides the next edit.** S1 withholds an apply that lands over an edit decided - after the flush mark, and a `Repair` puts the edit back. But the apply did land on the cluster, so - the next edit of that object carries the apply's value for every field it did not change, and the - window commits the whole object under that author's name: the overwritten edit is lost, and the - `Repair` then applies the loss. The failing test is - `TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit` - (`internal/git/apply_window_carry_test.go`, skipped; `GITOPS_REVERSER_KNOWN_BUGS=1` runs it). +Nothing open. ## Applier claims and bi-directional editing +- [ ] **Persist the withheld-field record.** The fields a withheld apply reverted are kept in memory + (`internal/git/withheld_fields.go`), so after a restart the next edit of such an object carries the + apply's values into Git again, until the `Repair` lands. - [ ] **Reopen the render-fidelity gate after a Git repair, automatically.** Today a target on `RenderMatchesLive=False` clears only on an explicit recheck (the `reconcile.configbutler.ai/requestedAt` annotation, which restarts every scope; the render diff --git a/docs/design/applier-echoes.md b/docs/design/applier-echoes.md index 1ae72e9a..6a7f30bc 100644 --- a/docs/design/applier-echoes.md +++ b/docs/design/applier-echoes.md @@ -606,7 +606,8 @@ mirrors and the applier applies. ## Decided while building -The build departs from the plan above in twelve places, decided on 2026-10-20: +The build departs from the plan above in thirteen places, the first twelve decided on 2026-10-20 and +the last on 2026-10-28: 1. **Siblings do not count for F3.** Whether Git is ahead, and the apply epoch, are per worker: an apply due on another worker writing the same branch does not hold this worker's resyncs. This @@ -672,6 +673,24 @@ The build departs from the plan above in twelve places, decided on 2026-10-20: attempt is sent for the tip that holds the siblings' writes. A sibling's push is foreign to the others, as it was when they found it by a fetch: its appliers may apply another cluster. A sibling's edit made while the apply runs stays unprotected, as it was before this change. +13. **A withheld apply's values are kept out of the next edit.** The plan withheld the apply and left + the rest to the `Repair`, but the apply did land: until the `Repair`, the cluster runs its values, + and the next event of the object, by anyone, carried each one its author did not change. A window + commits whole objects, so that author's commit wrote the apply's values, the edit was lost, and + the `Repair` applied the loss. The worker now keeps, per object, the last version an apply window + or a flush recorded as an edit (the last accepted version), and on a withhold records each field + where the withheld object differs from it: the accepted value and the applier's, either of which + may be absent. Every later event of the object, mirrored or not and the `Repair`'s own landing + included, and every resync snapshot of it, is restored against that record: a field still + holding the applier's value was carried over and is set back to the accepted value; a field + holding any other value was changed by someone and leaves the record; a delete clears it. This is + the desired view of + [`window-field-ownership-and-apply-hold.md`](window-field-ownership-and-apply-hold.md#convergence-is-a-separate-operation), + limited to the fields a withheld apply reverted: commits stay whole objects, and no window owns + fields. Three limits are accepted. A list is one field, so an edit to another item of a list the + apply reverted still carries the reverted item. An author setting a field to the applier's value + cannot be told from carrying it, and is set back. The record lives in memory, so a restart loses + it ([`withheld_fields.go`](../../internal/git/withheld_fields.go)). ## References diff --git a/internal/git/applier_claim.go b/internal/git/applier_claim.go index d6fa44df..eb9d2c06 100644 --- a/internal/git/applier_claim.go +++ b/internal/git/applier_claim.go @@ -402,6 +402,7 @@ func (l *branchWorkerEventLoop) serviceAppliers() { defer l.publishClaimStatus() defer l.publishHeld() now := time.Now() + defer l.forgetAccepted(now) l.settleApplyResults(now) l.followRecordedOperations(now) converging := l.scheduleConverges(now) @@ -660,7 +661,8 @@ func (l *branchWorkerEventLoop) endFlush() { // recordUnpublishedEdits records in the window of an attempt being sent the edits its commit does // not hold, which the apply may overwrite: those decided after the flush mark, and the live events -// held for a GitTarget's render fidelity. +// held for a GitTarget's render fidelity. Every live event of a held GitTarget is held, in order, so +// the last one held of an object is its last accepted version. func (l *branchWorkerEventLoop) recordUnpublishedEdits(state *applierState) { for _, target := range state.targets { for key, at := range l.flushEdits[target.ref] { @@ -669,6 +671,7 @@ func (l *branchWorkerEventLoop) recordUnpublishedEdits(state *applierState) { if held := l.held[target.ref]; held != nil { for _, h := range held.events { state.window.recordEdit(objectKey(h.event.Identifier), h.event.LastFieldManagedAt) + l.keepAccepted(h.event) } } } diff --git a/internal/git/apply_window.go b/internal/git/apply_window.go index dba110fa..603416bf 100644 --- a/internal/git/apply_window.go +++ b/internal/git/apply_window.go @@ -276,13 +276,15 @@ func (state *applierState) covers(event Event) bool { // and must not be mirrored: the apply itself landing on an object nobody edited (F1), or the apply // arriving late over a newer API edit (withheld, with a Repair pending). Every other event is // recorded as an API-side edit in the windows of the appliers that cover it, and in the edits the -// attempts of a flush in progress will hold once they are sent (recordFlushEdit). It is called for -// each live event before it reaches the commit window. +// attempts of a flush in progress will hold once they are sent (recordFlushEdit). An event recorded +// either way is kept as its object's last accepted version, what a withheld apply is compared with +// (withheld_fields.go). It is called for each live event before it reaches the commit window. func (l *branchWorkerEventLoop) withholdLateApply(event Event) bool { if len(l.appliers) == 0 { return false } now := time.Now() + recorded := false for _, ref := range l.sortedAppliers() { applier := l.appliers[ref] if !applier.covers(event) { @@ -293,6 +295,7 @@ func (l *branchWorkerEventLoop) withholdLateApply(event Event) bool { if l.judgeInWindow(applier, event, now) { return true } + recorded = true case event.Operation != string(itypes.OperationDelete): if _, alone := ref.writtenBy(event.LastFieldManagers); alone { // No apply this worker asked for is landing: an echo after the grace, or an apply @@ -304,6 +307,9 @@ func (l *branchWorkerEventLoop) withholdLateApply(event Event) bool { // Decided after the flush mark, so not in the commit the flush's attempts apply: their windows // hold it from the start (S1). l.recordFlushEdit(event) + if recorded || l.flushing { + l.keepAccepted(event) + } return false } @@ -351,8 +357,10 @@ func (l *branchWorkerEventLoop) judgeInWindow(applier *applierState, event Event // apply over the edit: events arrive in resourceVersion order, so the edit, recorded // already, was written first, and mirroring this would revert it in Git and on the // cluster. Only an undo in that same second is lost (C4 in docs/design/applier-echoes.md). - // An edit still held for render fidelity is repaired once it reaches Git. + // An edit still held for render fidelity is repaired once it reaches Git. The fields it + // reverted are recorded, so the next event of the object does not carry them into Git. l.recordApplyWindowCollision(applier, window, event, collisionWithheld) + l.recordWithheld(event) if !l.repairOnRelease(applier.ref, event) { applier.repair = true } diff --git a/internal/git/apply_window_carry_test.go b/internal/git/apply_window_carry_test.go index bd024da7..969c9a10 100644 --- a/internal/git/apply_window_carry_test.go +++ b/internal/git/apply_window_carry_test.go @@ -3,100 +3,253 @@ package git import ( - "os" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer" itypes "github.com/ConfigButler/gitops-reverser/internal/types" ) -// TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit is a known bug (docs/TODO.md, "A -// withheld apply rides the next edit"), written now and skipped until it is fixed; set -// GITOPS_REVERSER_KNOWN_BUGS=1 to run it. +// carryEdit is author's live update of y, holding both of its keys. +func carryEdit(author, a, b string) Event { + event := configMapTargetEvent("y", author, echoTarget) + event.Operation = string(itypes.OperationUpdate) + event.Object.Object["data"] = map[string]interface{}{"a": a, "b": b} + return event +} + +// carryApply is the applier writing y, after every edit so far. +func carryApply(a, b string) Event { + event := carryEdit("system:serviceaccount:flux-system:kustomize-controller", a, b) + event.LastFieldManagers = []string{FluxFieldManager} + event.LastFieldManagedAt = time.Now().Add(time.Hour) + return event +} + +// carryFlushOverAnEdit makes an apply due while x2 sits in the open window, and lets the flush +// publish it only after alice's edit a=a2 (E1) is decided: the attempt goes for a commit that lacks +// the edit, which Git holds as a1. +func carryFlushOverAnEdit(t *testing.T, slug string) *echoFixture { + t.Helper() + f := newEchoFixture(t, slug, configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) + f.writeWindow(carryEdit("alice", "a1", "b1")) // y as Git holds it before anyone edits + requester := f.withClaim() + holdPushes(f.loop) + + f.writeWindow(echoEvent("x", "x2")) + restore := failPushes(t, f.worker) + f.echoApplier().repair = true + f.loop.endWake(0) + + f.writeWindow(carryEdit("alice", "a2", "b1")) // E1, decided after the mark + restore() + f.loop.retry.due = time.Now().Add(-time.Second) + f.loop.runRetry() + f.loop.endWake(0) + require.Len(t, requester.taken(), 1, "the apply is sent without E1") + return f +} + +// publish closes the open window and pushes it. +func (f *echoFixture) publish() { + f.t.Helper() + f.loop.lastPushAt = time.Time{} + require.True(f.t, f.loop.finalizeOpenWindow()) + f.loop.pushPending() +} + +// carryWithheld publishes E1, then has the apply land over it, late: the apply is withheld, a Repair +// is pending, and the cluster runs a1 while Git holds a2. +func carryWithheld(t *testing.T, slug string) *echoFixture { + t.Helper() + f := carryFlushOverAnEdit(t, slug) + f.publish() + require.Contains(t, f.onMain(echoFile("y")), "a: a2", "precondition: E1 is published") + + f.writeWindow(carryApply("a1", "b1")) + require.True(t, f.echoApplier().repair, "the apply over E1 is withheld, and a Repair is pending") + require.Len(t, f.loop.withheld[carryKey()], 1, "the field the apply reverted is recorded") + return f +} + +// carryKey is y's objectKey. +func carryKey() string { return objectKey(carryEdit("alice", "", "").Identifier) } + +// TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit pins the withheld-field record +// (withheld_fields.go). // // S1 withholds an apply that lands over an edit decided after the flush mark, and a Repair puts the // edit back once it is published. But the apply did land on the cluster, and a window holds whole // objects: the next edit of that object, by the same author, carries the apply's value for every -// field it did not touch, and its commit writes that value under the author's name. The edit the -// apply overwrote is lost, and the Repair puts the loss back on the cluster. +// field it did not touch. Committed as it came, it wrote that value under the author's name, the +// edit the apply overwrote was lost, and the Repair put the loss back on the cluster. func TestApplyWindow_AWithheldApplyIsNotCommittedUnderTheNextEdit(t *testing.T) { - if os.Getenv("GITOPS_REVERSER_KNOWN_BUGS") == "" { - t.Skip("known bug, docs/TODO.md: a withheld apply rides the next edit") - } - - // edit is alice's live update of y, holding both of its keys. - edit := func(a, b string) Event { - event := configMapTargetEvent("y", "alice", echoTarget) - event.Operation = string(itypes.OperationUpdate) - event.Object.Object["data"] = map[string]interface{}{"a": a, "b": b} - return event - } - // lateApply is the applier writing y back to what the flushed commit holds, after the edit. - lateApply := func() Event { - event := edit("a1", "b1") - event.UserInfo.Username = "system:serviceaccount:flux-system:kustomize-controller" - event.LastFieldManagers = []string{FluxFieldManager} - event.LastFieldManagedAt = time.Now().Add(time.Hour) - return event - } - // flushOverAnEdit makes an apply due while x2 sits in the open window, and lets the flush publish - // it only after alice's edit a=a2 is decided: the attempt goes for a commit that lacks the edit. - flushOverAnEdit := func(t *testing.T, slug string) (*echoFixture, *recordingRequester) { - f := newEchoFixture(t, slug, configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) - f.writeWindow(edit("a1", "b1")) // y as Git holds it before anyone edits - requester := f.withClaim() - holdPushes(f.loop) - - f.writeWindow(echoEvent("x", "x2")) - restore := failPushes(t, f.worker) - f.echoApplier().repair = true - f.loop.endWake(0) - - f.writeWindow(edit("a2", "b1")) // E1, decided after the mark - restore() - f.loop.retry.due = time.Now().Add(-time.Second) - f.loop.runRetry() - f.loop.endWake(0) - require.Len(t, requester.taken(), 1, "the apply is sent without E1") - return f, requester - } - t.Run("in the window that holds the edit", func(t *testing.T) { - f, _ := flushOverAnEdit(t, "carry-same-window") + f := carryFlushOverAnEdit(t, "carry-same-window") require.NotNil(t, f.loop.openWindow, "E1 is collecting in alice's window") - f.writeWindow(lateApply()) + f.writeWindow(carryApply("a1", "b1")) require.True(t, f.echoApplier().repair, "the apply over E1 is withheld, and a Repair is pending") - f.writeWindow(edit("a1", "b2")) // E2: alice changes b on the object the apply wrote - f.loop.lastPushAt = time.Time{} - require.True(t, f.loop.finalizeOpenWindow()) - f.loop.pushPending() + f.writeWindow(carryEdit("alice", "a1", "b2")) // E2: alice changes b on the object the apply wrote + f.publish() require.Equal(t, "alice", gitIn(t, f.repoDir, "log", "-1", "--format=%an", "main")) assert.Contains(t, f.onMain(echoFile("y")), "a: a2", - "E1 is alice's, and the apply that overwrote it was withheld; Git loses it") + "E1 is alice's, and the apply that overwrote it was withheld; Git keeps it") + assert.Contains(t, f.onMain(echoFile("y")), "b: b2") }) t.Run("after the edit is published", func(t *testing.T) { - f, _ := flushOverAnEdit(t, "carry-next-window") - f.loop.lastPushAt = time.Time{} - require.True(t, f.loop.finalizeOpenWindow()) - f.loop.pushPending() - require.Contains(t, f.onMain(echoFile("y")), "a: a2", "precondition: E1 is published") - - f.writeWindow(lateApply()) - require.True(t, f.echoApplier().repair, "the apply over E1 is withheld, and a Repair is pending") - f.writeWindow(edit("a1", "b2")) - f.loop.lastPushAt = time.Time{} - require.True(t, f.loop.finalizeOpenWindow()) - f.loop.pushPending() + f := carryWithheld(t, "carry-next-window") + f.writeWindow(carryEdit("alice", "a1", "b2")) + f.publish() require.Equal(t, "alice", gitIn(t, f.repoDir, "log", "-1", "--format=%an", "main")) assert.Contains(t, f.onMain(echoFile("y")), "a: a2", - "alice's commit of b reverts her own E1 to the withheld apply's a1") + "alice's commit of b keeps her own E1, not the withheld apply's a1") + assert.Contains(t, f.onMain(echoFile("y")), "b: b2") + assert.Len(t, f.loop.withheld[carryKey()], 1, "the cluster still runs a1: the record stays") + }) + + t.Run("an edit of another author does not carry it either", func(t *testing.T) { + f := carryWithheld(t, "carry-other-author") + f.writeWindow(carryEdit("bob", "a1", "b2")) + f.publish() + + require.Equal(t, "bob", gitIn(t, f.repoDir, "log", "-1", "--format=%an", "main")) + assert.Contains(t, f.onMain(echoFile("y")), "a: a2", "bob changed b only") + assert.Contains(t, f.onMain(echoFile("y")), "b: b2") + }) +} + +// TestApplyWindow_TheWithheldRecordEnds pins when a field leaves the record: when an event shows it +// holding another value than the apply's, and when the object is deleted. +func TestApplyWindow_TheWithheldRecordEnds(t *testing.T) { + t.Run("an author who changes the field wins", func(t *testing.T) { + f := carryWithheld(t, "carry-author-wins") + f.writeWindow(carryEdit("alice", "a3", "b1")) // E2: alice changes a herself + f.publish() + + assert.Contains(t, f.onMain(echoFile("y")), "a: a3", "her value is not the apply's: it is hers") + assert.Empty(t, f.loop.withheld, "the field left the record") + }) + + t.Run("the Repair landing clears it", func(t *testing.T) { + f := carryWithheld(t, "carry-repair-lands") + f.writeWindow(carryApply("a2", "b1")) // the Repair puts E1 back on the cluster + assert.Empty(t, f.loop.withheld, "the cluster runs the accepted value again") + + f.writeWindow(carryEdit("alice", "a1", "b1")) // alice sets a1 herself, now + f.publish() + assert.Contains(t, f.onMain(echoFile("y")), "a: a1", "an edit after the Repair is not set back") + }) + + t.Run("a delete clears it", func(t *testing.T) { + f := carryWithheld(t, "carry-delete") + deleted := carryEdit("alice", "a1", "b1") + deleted.Operation = string(itypes.OperationDelete) + f.writeWindow(deleted) + assert.Empty(t, f.loop.withheld) + + created := carryEdit("alice", "a1", "b2") + created.Operation = string(itypes.OperationCreate) + f.writeWindow(created) + f.publish() + assert.Contains(t, f.onMain(echoFile("y")), "a: a1", "the object made again is mirrored as it is") }) } + +// TestApplyWindow_AResyncAfterAWithholdKeepsTheEdit pins the record on a resync: its snapshot holds +// what the cluster runs, the withheld apply's a1 included, and is set back as a live event is. +func TestApplyWindow_AResyncAfterAWithholdKeepsTheEdit(t *testing.T) { + f := carryWithheld(t, "carry-resync") + snapshot := carryEdit("alice", "a1", "b2") + req := &ResyncRequest{ + Desired: []manifestanalyzer.DesiredResource{ + {Resource: snapshot.Identifier, Object: snapshot.Object}, + echoDesired("x", "x2"), + }, + GitTargetName: echoTarget, + GitTargetNamespace: "default", + // A snapshot F3 lets through while the Repair is pending. + Initial: true, + Result: make(chan ResyncResult, 1), + } + f.loop.handleQueueItem(resyncItem(req)) + require.NoError(t, (<-req.Result).Err) + f.loop.lastPushAt = time.Time{} + f.loop.pushPending() + + assert.Contains(t, f.onMain(echoFile("y")), "a: a2", "the resync keeps E1") + assert.Contains(t, f.onMain(echoFile("y")), "b: b2", "and mirrors the rest of the snapshot") + assert.Equal(t, "a1", snapshot.Object.Object["data"].(map[string]interface{})["a"], + "the snapshot itself is left as it was") +} + +// TestWithheldFields_RecordAndRestore pins the record below the worker: the fields recorded are +// where the withheld object differs from the accepted one, a map key by key and a list as one field, +// with absence on either side; a later object is restored where it still holds the apply's value. +func TestWithheldFields_RecordAndRestore(t *testing.T) { + loop := &branchWorkerEventLoop{w: &BranchWorker{}} + accepted := carryEdit("alice", "", "") + accepted.Object.Object["data"] = map[string]interface{}{"a": "a2", "c": "c1"} + accepted.Object.Object["spec"] = map[string]interface{}{"list": []interface{}{"one", "two"}} + loop.keepAccepted(accepted) + + withheld := carryApply("", "") + withheld.Object.Object["data"] = map[string]interface{}{"a": "a1"} + withheld.Object.Object["spec"] = map[string]interface{}{"list": []interface{}{"one"}} + withheld.Object.Object["extra"] = "x" + loop.recordWithheld(withheld) + + key := carryKey() + assert.Equal(t, []withheldField{ + {Path: []string{"data", "a"}, Accepted: fieldValue{Value: "a2"}, Applier: fieldValue{Value: "a1"}}, + {Path: []string{"data", "c"}, Accepted: fieldValue{Value: "c1"}, Applier: fieldValue{Absent: true}}, + {Path: []string{"extra"}, Accepted: fieldValue{Absent: true}, Applier: fieldValue{Value: "x"}}, + { + Path: []string{"spec", "list"}, + Accepted: fieldValue{Value: []interface{}{"one", "two"}}, + Applier: fieldValue{Value: []interface{}{"one"}}, + }, + }, loop.withheld[key]) + + // A later object: b added, data.a carried, data.c still missing, extra changed by someone, the + // list carried. + later := withheld.Object.DeepCopy() + later.Object["data"].(map[string]interface{})["b"] = "b2" + later.Object["extra"] = "y" + restored := loop.restoreWithheld(key, later) + + require.NotSame(t, later, restored, "the object set back is a copy") + assert.Equal(t, map[string]interface{}{"a": "a2", "b": "b2", "c": "c1"}, restored.Object["data"]) + assert.Equal(t, "y", restored.Object["extra"], "a field someone changed is theirs") + assert.Equal(t, []interface{}{"one", "two"}, restored.Object["spec"].(map[string]interface{})["list"]) + assert.Len(t, loop.withheld[key], 3, "extra left the record") + + // A second withhold, over a version without data at all, replaces the fields inside it. + loop.keepAccepted(Event{Identifier: accepted.Identifier, Object: restored}) + again := carryApply("", "") + delete(again.Object.Object, "data") + again.Object.Object["spec"] = map[string]interface{}{"list": []interface{}{"one"}} + again.Object.Object["extra"] = "y" + loop.recordWithheld(again) + paths := make([][]string, 0, len(loop.withheld[key])) + for _, field := range loop.withheld[key] { + paths = append(paths, field.Path) + } + assert.ElementsMatch(t, [][]string{{"data"}, {"spec", "list"}}, paths) + + // An object that changed nothing the record holds is returned as it is, and ends the record. + unrelated := &unstructured.Unstructured{Object: map[string]interface{}{ + "data": map[string]interface{}{"a": "a9"}, + "spec": map[string]interface{}{"list": []interface{}{"three"}}, + }} + assert.Same(t, unrelated, loop.restoreWithheld(key, unrelated)) + assert.Empty(t, loop.withheld) +} diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index 65c73fb7..39b0f1a3 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -22,6 +22,7 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -1107,6 +1108,11 @@ type branchWorkerEventLoop struct { flushing bool flushMark uint64 flushEdits map[itypes.ResourceReference]map[string]time.Time + // accepted is, by objectKey, the object of the last event an open apply window or the flush + // recorded as an edit, and withheld the fields withheld applies reverted that no event has + // changed since: the next event of the object is restored against them (withheld_fields.go). + accepted map[string]*unstructured.Unstructured + withheld withheldFields // attachTimer fires at the earliest attach deadline of a waiting CommitRequest, so a request // that no window reaches still resolves on time. An attached request needs no timer of its own: @@ -1305,7 +1311,7 @@ func (l *branchWorkerEventLoop) dispatch(item WorkItem) int64 { // finalizing it as identity, the byte cap or a zero window require. func (l *branchWorkerEventLoop) handleLiveEvents(request *WriteRequest) { for _, event := range request.Events { - if !l.mirrors(event) { + if !l.mirrors(&event) { continue } l.w.liveEdits.Add(1) @@ -1390,7 +1396,11 @@ func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { // divergence the event is dropped, and while the gate is pending it is held, once judged as on // arrival, and committed in order when the gate opens (render_fidelity_hold.go). Events held for the // target earlier go first, once its gate opened. -func (l *branchWorkerEventLoop) mirrors(event Event) bool { +// +// Every event is first restored against the fields withheld applies reverted, mirrored or not, so +// the event carries none of them into a commit (withheld_fields.go). +func (l *branchWorkerEventLoop) mirrors(event *Event) bool { + l.restoreWithheldEvent(event) target := itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace) l.serviceHeld(target) status, closed := l.w.renderFidelityClosed(event.GitTargetName, event.GitTargetNamespace) @@ -1404,16 +1414,16 @@ func (l *branchWorkerEventLoop) mirrors(event Event) bool { } // Judged when it arrives, held or not: the apply windows and the flush record an edit when it // was made. - if l.withholdLateApply(event) { + if l.withholdLateApply(*event) { return false } if closed || l.held[target] != nil { // Its adoption waits with it (F-c): an apply before the create is in a commit stamps nothing. - l.holdAtRenderFidelity(target, event, status) + l.holdAtRenderFidelity(target, *event, status) return false } if event.Operation == string(itypes.OperationCreate) { - l.noteCreated(event) + l.noteCreated(*event) } return true } diff --git a/internal/git/resync_flush.go b/internal/git/resync_flush.go index 5ee78aee..066aaba1 100644 --- a/internal/git/resync_flush.go +++ b/internal/git/resync_flush.go @@ -57,6 +57,9 @@ func (l *branchWorkerEventLoop) applyResync(req *ResyncRequest) { return } pendingWrite.Desired, pendingWrite.ownedElsewhere = l.withoutOwnedElsewhere(req) + // The snapshot holds what the cluster runs, a withheld apply's values included: they are set + // back as in a live event (withheld_fields.go). + pendingWrite.Desired = l.restoreWithheldSnapshot(pendingWrite.Desired) pendingWrite.Committed = new(bool) pendingWrite.origin.resync = req l.decide(*pendingWrite) diff --git a/internal/git/withheld_fields.go b/internal/git/withheld_fields.go new file mode 100644 index 00000000..779642a0 --- /dev/null +++ b/internal/git/withheld_fields.go @@ -0,0 +1,314 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "maps" + "reflect" + "slices" + "time" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + + "github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer" + itypes "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// A withheld apply (apply_window.go) is not mirrored, but it did land: the cluster runs the values it +// reverted until a Repair puts the edit back. A window holds whole objects, so the next event of the +// object, by anyone, carries each reverted value its author did not change. Committed as it came, it +// wrote the apply's value under that author's name, the edit the apply overwrote was lost, and the +// Repair then applied the loss. +// +// So the worker records, per object, the fields a withheld apply reverted: what the object's last +// accepted version held there, and what the apply wrote. The last accepted version is the object of +// the last event an apply window or a flush recorded as an edit (keepAccepted). This is the desired +// view of docs/design/window-field-ownership-and-apply-hold.md, limited to the fields a withheld apply +// reverted; commits stay whole objects, and no window owns fields. +// +// Every later event of the object, mirrored or not (the Repair's own landing included), and every +// snapshot of it a resync decides, is restored against the record: a field still holding the apply's +// value was carried over, and is set back to the accepted value; a field holding any other value was +// changed by someone since, and leaves the record. A delete clears the object's record. +// +// Decided limits: a list is one field, so an edit to another item of a list the apply reverted still +// carries the reverted item; an author setting a field to the apply's value cannot be told from +// carrying it, and is set back; and the record lives in memory, so a restart loses it. + +// withheldFields is, by objectKey, the fields withheld applies reverted that no event has changed +// since. It is plain data. +type withheldFields map[string][]withheldField + +// withheldField is one field a withheld apply reverted. +type withheldField struct { + // Path is the field's map keys from the object's root. A list is one field: a path never goes + // into one. + Path []string + // Accepted is what the object's last accepted version held at Path, Applier what the withheld + // apply wrote there. + Accepted fieldValue + Applier fieldValue +} + +// fieldValue is a field's value, or its absence. +type fieldValue struct { + Value any + Absent bool +} + +// equal reports whether two field values are the same value, or both absent. +func (v fieldValue) equal(other fieldValue) bool { + return v.Absent == other.Absent && (v.Absent || reflect.DeepEqual(v.Value, other.Value)) +} + +// keepAccepted keeps a live event's object as its object's last accepted version, after an apply +// window or a flush recorded the event as an edit. A delete, and a field patch, which carries no +// whole object, forget it: with no version to compare with, a withhold records nothing. +func (l *branchWorkerEventLoop) keepAccepted(event Event) { + key := objectKey(event.Identifier) + if event.Object == nil || event.Operation == string(itypes.OperationDelete) { + delete(l.accepted, key) + return + } + if l.accepted == nil { + l.accepted = map[string]*unstructured.Unstructured{} + } + l.accepted[key] = event.Object.DeepCopy() +} + +// forgetAccepted drops the last accepted versions of the objects no open apply window and no flush +// records an edit of: a write to one of those is never withheld (judgeInWindow). +func (l *branchWorkerEventLoop) forgetAccepted(now time.Time) { + for key := range l.accepted { + if !l.editRecorded(key, now) { + delete(l.accepted, key) + } + } +} + +// editRecorded reports whether an open apply window, or the flush in progress, records an edit of +// the object. +func (l *branchWorkerEventLoop) editRecorded(key string, now time.Time) bool { + for _, state := range l.appliers { + window := state.window + if !window.open(now) { + continue + } + if _, edited := window.edited[key]; edited { + return true + } + if _, carried := window.carried[key]; carried && window.carrying(now) { + return true + } + } + for _, edits := range l.flushEdits { + if _, edited := edits[key]; edited { + return true + } + } + return false +} + +// recordWithheld records the fields a withheld apply reverted: where its object differs from the +// object's last accepted version. A field recorded already is replaced, with any recorded inside it +// or around it. With no accepted version kept there is nothing to compare with, and nothing is +// recorded. +// +// The event was restored first (restoreWithheld), so a field it still carried from an earlier +// withheld apply holds the accepted value, differs in nothing, and keeps its record. +func (l *branchWorkerEventLoop) recordWithheld(event Event) { + key := objectKey(event.Identifier) + accepted := l.accepted[key] + if accepted == nil || event.Object == nil { + return + } + reverted := diffFields(nil, accepted.Object, event.Object.Object) + if len(reverted) == 0 { + return + } + fields := slices.DeleteFunc(slices.Clone(l.withheld[key]), func(field withheldField) bool { + return slices.ContainsFunc(reverted, func(other withheldField) bool { return overlaps(field.Path, other.Path) }) + }) + if l.withheld == nil { + l.withheld = withheldFields{} + } + l.withheld[key] = append(fields, reverted...) + l.w.Log.V(1).Info("Recorded the fields a withheld apply reverted, so the next event of the object does "+ + "not carry them into Git", "branch", l.w.Branch, "object", event.Identifier.String(), "fields", len(reverted)) +} + +// restoreWithheldEvent restores a live event against the record of its object, before it is +// judged: a delete clears the record, and any other event with a whole object has each field it +// carried from a withheld apply set back (restoreWithheld). +func (l *branchWorkerEventLoop) restoreWithheldEvent(event *Event) { + key := objectKey(event.Identifier) + if event.Operation == string(itypes.OperationDelete) { + delete(l.withheld, key) + return + } + if restored := l.restoreWithheld(key, event.Object); restored != event.Object { + l.w.Log.Info("Set back the fields a withheld apply reverted, which the event carried: the cluster "+ + "still runs the apply's values, and the event's author did not change them", + "branch", l.w.Branch, "object", event.Identifier.String(), "author", event.UserInfo.Username) + event.Object = restored + } +} + +// restoreWithheldSnapshot restores each object of a resync's snapshot against its record, as a live +// event is. The snapshot itself is left as it was: an object set back is a copy, in a copy of it. +func (l *branchWorkerEventLoop) restoreWithheldSnapshot( + desired []manifestanalyzer.DesiredResource, +) []manifestanalyzer.DesiredResource { + if len(l.withheld) == 0 { + return desired + } + var restored []manifestanalyzer.DesiredResource + for i, dr := range desired { + object := l.restoreWithheld(objectKey(dr.Resource), dr.Object) + if object == dr.Object { + continue + } + if restored == nil { + restored = slices.Clone(desired) + } + restored[i].Object = object + } + if restored == nil { + return desired + } + return restored +} + +// restoreWithheld applies the record of an object to one of its later versions, and updates the +// record: a field still holding the withheld apply's value is set back to the accepted value, and +// stays recorded; a field holding any other value leaves the record. It returns the object to +// mirror, a copy when a field was set back. +func (l *branchWorkerEventLoop) restoreWithheld( + key string, object *unstructured.Unstructured, +) *unstructured.Unstructured { + fields := l.withheld[key] + if len(fields) == 0 || object == nil { + return object + } + restored := object + var kept []withheldField + for _, field := range fields { + value, ok := lookupField(object.Object, field.Path) + if !ok || !value.equal(field.Applier) { + continue + } + kept = append(kept, field) + if restored == object { + restored = object.DeepCopy() + } + setField(restored.Object, field.Path, field.Accepted) + } + if len(kept) == 0 { + delete(l.withheld, key) + } else { + l.withheld[key] = kept + } + return restored +} + +// diffFields lists the fields where applier differs from accepted, under path. Two maps are compared +// key by key; anything else, a list included, is one field. +func diffFields(path []string, accepted, applier map[string]any) []withheldField { + var fields []withheldField + keys := slices.Collect(maps.Keys(accepted)) + for key := range applier { + if _, both := accepted[key]; !both { + keys = append(keys, key) + } + } + slices.Sort(keys) + for _, key := range keys { + at := append(slices.Clone(path), key) + was, inAccepted := accepted[key] + now, inApplier := applier[key] + wasMap, wasIsMap := was.(map[string]any) + nowMap, nowIsMap := now.(map[string]any) + switch { + case inAccepted && inApplier && wasIsMap && nowIsMap: + fields = append(fields, diffFields(at, wasMap, nowMap)...) + case inAccepted && inApplier && reflect.DeepEqual(was, now): + default: + fields = append(fields, withheldField{ + Path: at, + Accepted: fieldValue{Value: copyValue(was), Absent: !inAccepted}, + Applier: fieldValue{Value: copyValue(now), Absent: !inApplier}, + }) + } + } + return fields +} + +// lookupField is the value at path in object, or its absence. It is not ok when the path runs +// through something that is not a map: whatever is there, it is not what the record compares. +func lookupField(object map[string]any, path []string) (fieldValue, bool) { + current := object + for i, key := range path { + value, found := current[key] + if !found { + return fieldValue{Absent: true}, true + } + if i == len(path)-1 { + return fieldValue{Value: value}, true + } + next, isMap := value.(map[string]any) + if !isMap { + return fieldValue{}, false + } + current = next + } + return fieldValue{}, false +} + +// setField sets the field at path in object to value, or removes it when value is absent. A map on +// the way that is missing is created. +func setField(object map[string]any, path []string, value fieldValue) { + current := object + for _, key := range path[:len(path)-1] { + next, isMap := current[key].(map[string]any) + if !isMap { + if value.Absent { + return + } + next = map[string]any{} + current[key] = next + } + current = next + } + last := path[len(path)-1] + if value.Absent { + delete(current, last) + return + } + current[last] = copyValue(value.Value) +} + +// overlaps reports whether one path is the other or lies inside it. +func overlaps(a, b []string) bool { + n := min(len(a), len(b)) + return slices.Equal(a[:n], b[:n]) +} + +// copyValue deep-copies a value of an unstructured object. +func copyValue(value any) any { + switch typed := value.(type) { + case map[string]any: + copied := make(map[string]any, len(typed)) + for key, item := range typed { + copied[key] = copyValue(item) + } + return copied + case []any: + copied := make([]any, len(typed)) + for i, item := range typed { + copied[i] = copyValue(item) + } + return copied + default: + return value + } +}