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 + } +}