diff --git a/api/v1alpha3/gitbranch_types.go b/api/v1alpha3/gitbranch_types.go index dbce71bc..ef557fa2 100644 --- a/api/v1alpha3/gitbranch_types.go +++ b/api/v1alpha3/gitbranch_types.go @@ -58,9 +58,9 @@ type GitBranchSpec struct { // AllowedGitTargets says which GitTargets may write through this GitBranch. Omitted, it defaults // to one grant for this GitBranch's own namespace (consumerNamespaces: {from: Same}). A list you - // write is the whole policy: nothing is added to it, so a list without a from: Same grant shuts - // this namespace out, and [] admits no GitTarget. Grants combine by union. Mutable: claiming a - // branch includes the right to share it. + // write is the whole policy: nothing is added to it, so this namespace is admitted only if the + // list grants it access, and [] admits no GitTarget. Grants combine by union. Mutable: claiming + // a branch includes the right to share it. // +optional // +listType=atomic // +kubebuilder:validation:MaxItems=64 diff --git a/config/crd/bases/configbutler.ai_gitbranches.yaml b/config/crd/bases/configbutler.ai_gitbranches.yaml index de025fd0..2acbfc8c 100644 --- a/config/crd/bases/configbutler.ai_gitbranches.yaml +++ b/config/crd/bases/configbutler.ai_gitbranches.yaml @@ -88,9 +88,9 @@ spec: description: |- AllowedGitTargets says which GitTargets may write through this GitBranch. Omitted, it defaults to one grant for this GitBranch's own namespace (consumerNamespaces: {from: Same}). A list you - write is the whole policy: nothing is added to it, so a list without a from: Same grant shuts - this namespace out, and [] admits no GitTarget. Grants combine by union. Mutable: claiming a - branch includes the right to share it. + write is the whole policy: nothing is added to it, so this namespace is admitted only if the + list grants it access, and [] admits no GitTarget. Grants combine by union. Mutable: claiming + a branch includes the right to share it. items: description: |- GitBranchGitTargetGrant admits GitTargets in some control-cluster namespaces to write through a diff --git a/docs/TODO.md b/docs/TODO.md index 705c85a7..6c4fda5f 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -3,71 +3,41 @@ The open work, in one list. Each item is one or two lines with a code pointer or the design page that argues it. Ranked within each group, most urgent first. **Before the next release** marks what should land before the next tag is cut (the last release is v0.53.0). Checked against `main` at -`1c4fade4` on 2026-10-24. +`05799999` plus #443 on 2026-10-24. ## Before the next release -- [ ] **R1: an applier that comes due during a flush records nothing between the mark and its - joining.** Its `flushEdits` starts when it joins, not at the mark, so an edit decided in between - is neither published before the apply nor recorded; if the apply overwrites it, Git keeps the edit - and the cluster runs the old value with no `Repair` pending. Start `flushEdits` for every applier - when the flush begins, or take a new mark when the due set grows. `flushed()` in - `internal/git/applier_claim.go`. -- [ ] **R4: a replaced branch worker reads the old worker's settled-snapshot record.** - `forgetSettledSnapshots` runs only on `GitTarget` teardown (`internal/watch/materialization.go`), - so after "Branch worker replaced" (`gittarget_controller.go`, the `RequestRecheckForGitTarget` - path) the new worker's startup converge passes the barrier at once and can apply before the folder - is rebuilt. Argo CD can then be asked for a path that does not exist. -- [ ] **Override #428's changelog entry.** Its `BREAKING CHANGE` footer describes - `GitProvider.spec.accessFrom` and "every GitTarget writing one repository branch must name the same - GitProvider", both replaced before release by `GitBranch` and grants (#429, #441). Add a - `BEGIN_COMMIT_OVERRIDE` block to #428's PR body, as #419 did, before the release PR is cut. -- [ ] **Remove the "loud rejection" fields.** `RemovedReceiver` (`gitprovider_types.go`), - `RemovedClusterProviderRef` (`gittarget_types.go`, `fluxclaim_types.go`, `argocdclaim_types.go`), - `RemovedGitProviderRef`, `RemovedBranch` and `RemovedParentBranch` (`gittarget_types.go`) exist - only to refuse an old spelling at admission. With no users yet, an API break is a hard cut; drop - them and their CEL rules, and the matching sentences in [UPGRADING.md](UPGRADING.md). -- [ ] **F-i: CRD text beyond the contract.** `RepairPending` says "an API edit made after its - flush", and `kubectl explain` never defines "flush" (`fluxclaim_types.go`, `argocdclaim_types.go`). - Check `applierClaimRef`'s description for rationale too (AGENTS.md, API changes). +- [ ] **Override the changelog entries of #441 and #419.** Like #428's (overridden), #441's squash + footer narrates moves between shapes that never shipped (`from`/`names`/`selector` into + `consumerNamespaces`, `branches` renamed `branchPatterns`), and #419's override still names + `FluxTrigger`, `ArgoCDTrigger` and `spec.applyTriggerRef`, which #438 renamed. Give each a + `BEGIN_COMMIT_OVERRIDE` block that describes what ships, consistent with + [UPGRADING.md](UPGRADING.md), before the release PR is cut. ## Applier claims and bi-directional editing -- [ ] **Decide Fix 2 for live edits dropped while render fidelity is pending**: hold the events - (2a, the user stays the author) or re-derive them with a resync (2b, the operator becomes the - author), or keep the drop and alert on `git_writes_dropped_total`. It is also the cause behind the - flaky audit-route override e2e spec. - [live-edit-dropped-while-render-fidelity-pending.md](design/live-edit-dropped-while-render-fidelity-pending.md). -- [ ] **Reopen the render-fidelity gate after a Git repair.** A target stuck on - `RenderMatchesLive=False` clears only on a re-declaration or a restart. `Invalidate` and - `RequestRecheckForGitTarget` exist; their only caller is branch-worker replacement - (`gittarget_controller.go`). Specified in - [render-fidelity.md](design/support-boundary/render-fidelity.md) §6b; no decision needed. +- [ ] **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 + fidelity e2e spec proves it recovers a repaired folder), a re-declaration or a restart. Doing it on + an incoming Git revision is more than calling `Invalidate` and `RequestRecheckForGitTarget` from a + new place: it first needs the retained-intent ordering barrier, so the source refresh cannot + discard or race an open commit window or pending live edits + ([render-fidelity.md](design/support-boundary/render-fidelity.md) §6b). - [ ] **Watch the held applier**, so a switch-on or a `flux resume` is undone at once instead of within the one-minute `heldClaimSteadyInterval`. Also makes ask #31 below exact. [applier-claim-held-applier-watch.md](design/applier-claim-held-applier-watch.md). -- [ ] **F-a: a held barrier is invisible.** A `GitTarget` with no planned collection holds its - applier, and with it every reason of every `GitTarget` on that claim, `ForeignCommit` included, - with no log, condition or metric. The hold is approved; the silence is the defect. - (`internal/git/applier_claim.go`, `internal/watch/apply_state.go`) -- [ ] **F-e: the apply window compares two clocks.** `sentAt` is the operator's clock, - `LastFieldManagedAt` the source API server's, at second granularity; a skew of 1s flips the echo - decision. Add a skewed-clock test, and probably a one-second grace. (`internal/git/apply_window.go`) -- [ ] **F-b, F-c, F-d: edge cases of the snapshot hold.** Live events dropped at the gate while a - discarded snapshot is held (F-b); an adopt lost while render fidelity is closed, because - `mirrors()` returns before `noteCreated` (F-c); a `GitTarget` that names a claim after its - snapshots settled empty never gets its folder (F-d, `resync_flush.go`). +- [ ] **F3: a Flux fetch is made before the hold is checked**, so a `Kustomization` resumed by hand + applies the fetched artifact at once (`internal/applier/flux.go`, `Request`). +- [ ] **The apply window's remaining clock limits** (`internal/git/apply_window.go`, file comment). + A server clock ahead by more than the time between an earlier apply and the send dates an undo of + an edit published before the send after it, so the undo is not mirrored and the next apply reverts + it. With an edit carried from the window before, the check against the send still mixes the two + clocks, and a server clock behind can withhold this attempt's own apply and ask for an unneeded + `Repair`. Only comparing the event's content with Git separates them. - [ ] **F-h: small leaks and races.** The 5-minute hold can restart; the `Initial` flag races settlement; `snapshotHeldSince`, `applierState.named` and dropped collections in `settledSnapshots` leak. -- [ ] **F2: accept a Flux fetch whose token was stamped over** when the artifact holds the commit. - Today `Flux.fetch` waits only for its own token, so a commit another stamp fetched fails after 30s - (`internal/applier/flux.go`). Also check F3: the hold is checked after the fetch, so a - `Kustomization` resumed by hand applies the fetch at once. -- [ ] **A2: judge a gone Argo CD token by `status.history`** (our commit, - `initiatedBy: gitops-reverser`, `deployStartedAt` at or after our request means `Succeeded`). - Today a manual sync that wipes our token after a success reads as failed and syncs again, so hooks - run twice (`internal/applier/argocd.go`). - [ ] **Check Argo CD sync windows in `Configured`.** The guide says a sync window must not block GitOps Reverser's syncs; nothing checks it. - [ ] **Refuse or flag overrides that are not in Git** (Argo `spec.source.kustomize`; Flux @@ -220,6 +190,16 @@ Their argument for each is in [gitops-api-open-asks.md](future/gitops-api-open-a ## Docs and vocabulary +- [ ] **Cleanup: drop the "loud rejection" fields.** `RemovedReceiver` (`gitprovider_types.go`), + `RemovedClusterProviderRef` (`gittarget_types.go`, `fluxclaim_types.go`, `argocdclaim_types.go`), + `RemovedGitProviderRef`, `RemovedBranch` and `RemovedParentBranch` (`gittarget_types.go`) refuse an + old spelling at admission with a message naming its replacement. Keeping that message is not a + release defect; with no users yet they can go, with their CEL rules and the matching sentences in + [UPGRADING.md](UPGRADING.md), whenever convenient. +- [ ] **Cleanup, F-i: CRD text beyond the contract.** `RepairPending` says "an API edit made after + its flush", and `kubectl explain` never defines "flush" (`fluxclaim_types.go`, + `argocdclaim_types.go`). Check `applierClaimRef`'s description for rationale too (AGENTS.md, API + changes). - [ ] **Stale design-doc headers**, one docs PR: [open-asks-priority.md](design/open-asks-priority.md) (#339 closed, #23 fixed), [gittarget-api-wave.md](design/gittarget-api-wave.md) (riders shipped), [git-write-preflight.md](design/git-write-preflight.md) (superseded by the red-status plan), diff --git a/docs/UPGRADING.md b/docs/UPGRADING.md index c895d68b..795b81ba 100644 --- a/docs/UPGRADING.md +++ b/docs/UPGRADING.md @@ -10,7 +10,7 @@ We are pre-1.0, so breaking changes bump the **minor** version (release-please i ## Upgrading from 0.53.0: recreate the connection objects in one pass **Breaking, with no in-place conversion.** The objects that connect a folder to a cluster, a -repository and an applier are reshaped, and the next five entries say how. `ClusterProvider` is a +repository and an applier are reshaped, and the entries below say how. `ClusterProvider` is a namespaced `KubeProvider`, a repository branch is a `GitBranch`, the push receiver is a `GitReceiver`, `FluxTrigger` and `ArgoCDTrigger` are `FluxClaim` and `ArgoCDClaim`, and grants say who may use each provider and branch. Every `GitTarget` and applier claim is recreated, because @@ -31,10 +31,14 @@ the new object takes it over on its first snapshot. `Application` keeps automated sync off, as you set it for 0.53.0, so Argo CD syncs it only when someone starts a sync until step 6. Then delete their CRDs: `kubectl delete crd fluxtriggers.configbutler.ai argocdtriggers.configbutler.ai`. -3. Upgrade the CRDs, the operator and its permissions. The chart's `clusterProvider.*` values are - `kubeProvider.*`, and the provider it renders is a `KubeProvider` named `local` in the release - namespace, which admits only the release namespace's own objects until you set its grants (see - the grants entry). +3. Upgrade the CRDs, the operator and its permissions. Remove `clusterProvider` from your Helm + values first: the chart's schema rejects it. `clusterProvider.createDefault` is + `kubeProvider.create`, `clusterProvider.default.kubeConfig.secretRef` is `kubeProvider.secretRef`, + and `accessFrom` and `allowCrossNamespace` are `kubeProvider.allowedGitTargets` and + `kubeProvider.allowedApplierClaims` (see the grants entry). The provider the chart renders is a + `KubeProvider` named `local` (the default of `kubeProvider.name`; use yours where you set it) in + the release namespace, where 0.53.0's was the `ClusterProvider` + `default`, and it admits only the release namespace's own objects until you set its grants. 4. Create the new connection objects: - a `KubeProvider` for each other `ClusterProvider`, in the namespace that should own the credential, with its kubeconfig Secret copied into that namespace; @@ -44,10 +48,15 @@ the new object takes it over on its first snapshot. - a `GitBranch` for each repository branch your `GitTarget`s wrote, from their `gitProviderRef`, `branch` and `parentBranch`. 5. Delete each `GitTarget` and create it again with `spec.gitBranchRef` and `spec.kubeProviderRef`, - and `spec.applierClaimRef` where it had `spec.applyTriggerRef`. -6. Create a `FluxClaim` or `ArgoCDClaim` for each trigger you deleted: the same `spec`, with - `spec.kubeProviderRef` for `spec.clusterProviderRef` and without `spec.quietPeriod`. It takes - hold of its applier again once a `GitTarget` names it. + and `spec.applierClaimRef` where it had `spec.applyTriggerRef`. `spec.kubeProviderRef` is + required: a `GitTarget` whose `spec.clusterProviderRef` named `default`, which 0.53.0 filled in + when it was omitted, now names the chart's provider, + `{name: local, namespace: }` (your `kubeProvider.name` in place of `local` where + you set one), which admits a `GitTarget` in another + namespace only through a grant. +6. Create a `FluxClaim` or `ArgoCDClaim` for each trigger you deleted: the same `spec`, with the + required `spec.kubeProviderRef` in place of `spec.clusterProviderRef`, named as in step 5, and + without `spec.quietPeriod`. It takes hold of its applier again once a `GitTarget` names it. 7. Point each API server's audit webhook at its provider's new route, or set `spec.attribution.auditRoute` to the route it posts under today. 8. Delete the old CRD, which Helm leaves installed: `kubectl delete crd clusterproviders.configbutler.ai`. @@ -211,7 +220,7 @@ What 0.53.0 allowed, written as grants: a `GitTarget` wrote any `allowedBranches ```yaml kind: GitProvider spec: - allowedGitBranches: # was allowedBranches: [main]; required, no default + allowedGitBranches: # was allowedBranches: [main]; no default: omitted admits nobody - consumerNamespaces: {from: Same} branchPatterns: [main] --- @@ -254,8 +263,9 @@ spec: reports `Validated=False` (`NamespaceNotAuthorized`). - **The chart's `local` provider admits only the release namespace** unless `kubeProvider.allowedGitTargets` and `kubeProvider.allowedApplierClaims` say otherwise; `[]` there - admits nobody. `quickstart.enabled=true` replaces the default with a grant for exactly the - quickstart namespace. To keep 0.53.0's open default, every namespace mirroring its own, install + admits nobody. `quickstart.enabled=true` adds a grant for exactly the quickstart namespace; with + the lists unset, that grant replaces the default rather than joining it. To keep 0.53.0's open + default, every namespace mirroring its own, install with the chart's `values-playground.yaml`, which sets the `KubeProvider` grants above. - **The default trusts the object's own namespace.** Whoever may create a `GitTarget` or an applier claim beside a `KubeProvider` may use its credential within the default scope. Keep a provider in diff --git a/docs/api-first-publication.md b/docs/api-first-publication.md index 64614b7f..7775d79d 100644 --- a/docs/api-first-publication.md +++ b/docs/api-first-publication.md @@ -180,7 +180,7 @@ them is retried automatically, so it is worth knowing which one you are looking | --- | --- | --- | | Sensitive-resource encryption | Secrets and configured sensitive types must never reach Git in plaintext | The write fails rather than falling back to plaintext. There is no opt-out | | Acceptance of the Git path | The target's folder must be content the writer can edit safely | The plan is refused and reported as `GitPathAccepted=False` on the `GitTarget`. The periodic refresh raises the same condition off a READ, so folder content somebody else pushed is reported before a write goes looking for it. The target re-checks roughly every ten seconds; recovery requires an accepted resync after correcting the folder or live state, or — for a refusal a read raised — a later read that finds the folder clean | -| Render fidelity | A target whose render-vs-live epoch is pending or divergent must not take live writes | Live events and atomic writes are dropped while the gate is closed, counted by `gitopsreverser_git_writes_dropped_total`, and the first drop under each gate status is logged at Info. Resync stays allowed so it can measure and repair Git. The gate is per `GitTarget`: while any of its collections is replaying, edits to all of them are dropped. A live edit queued behind a replay is not: the worker records the replay's result before it takes the next item | +| Render fidelity | A target whose render-vs-live epoch is pending or divergent must not take live writes | While the gate is pending (a collection of the `GitTarget` is replaying, which closes all of them), live events are held and committed in order, with their authors, once it opens; a replay of their collection supersedes the ones held before it, except a deletion of an object it does not hold. The hold gives up after 10 minutes, or past the branch's retained-byte budget, by dropping what it held and asking for a fresh snapshot. A divergence drops live events, and atomic writes are dropped while the gate is closed. Every drop is counted by `gitopsreverser_git_writes_dropped_total`, and the first under each gate status is logged at Info. Resync stays allowed so it can measure and repair Git | A refusal is not a transient error and is not retried into success. It is the common reason a live edit never appears in Git while the worker looks healthy, so check the `GitTarget` conditions before diff --git a/docs/configuration.md b/docs/configuration.md index d46e78a6..6b021fc0 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -298,7 +298,7 @@ object that lends something, and is named `allowed` plus the kind that refers to | You write | The list is | |---|---| | nothing | `GitBranch`: `[{consumerNamespaces: {from: Same}}]`. `KubeProvider`: `allowedGitTargets: [{consumerNamespaces: {from: Same}, sourceNamespaces: ["{consumer.namespace}"]}]` and `allowedApplierClaims: [{consumerNamespaces: {from: Same}, applierNamespaces: ["{consumer.namespace}"]}]` | - | a list | exactly that list; the default is not added, so a list without a `from: Same` grant shuts the own namespace out | + | a list | exactly that list; the default is not added, so the local namespace is admitted only if the list grants it access | | `[]` | nobody, the own namespace included | The default grants no cluster-scoped resource, and it reaches only the namespace of the same diff --git a/docs/design/live-edit-dropped-while-render-fidelity-pending.md b/docs/design/live-edit-dropped-while-render-fidelity-pending.md index f6e4d9de..6cfd0e79 100644 --- a/docs/design/live-edit-dropped-while-render-fidelity-pending.md +++ b/docs/design/live-edit-dropped-while-render-fidelity-pending.md @@ -1,11 +1,18 @@ # A live edit dropped while a GitTarget's watches start -> **Status (2026-10-24): Fix 1 is built, Fix 2 needs a decision.** Fix 1 and the drop counter -> shipped with #420 (d7ce1bd4): the worker records a scope clean itself (`recordReplayClean` in +> **Status (2026-10-24): Fix 1 and Fix 2a are built.** Fix 1 and the drop counter shipped with #420 +> (d7ce1bd4): the worker records a scope clean itself (`recordReplayClean` in > `internal/git/render_fidelity_drops.go`), and `gitopsreverser_git_writes_dropped_total{kind, -> reason}` counts every drop, the first per gate status logged at Info. What is left is -> [Fix 2](#fix-2-stop-dropping-live-events-while-the-gate-is-pending): an edit that arrives while -> another collection of the same `GitTarget` is still replaying is still dropped. +> reason}` counts every drop. [Fix 2a](#2a-hold-the-events-release-them-when-the-gate-opens) is +> built in `internal/git/render_fidelity_hold.go`: a live event that meets a pending gate is held and +> committed with its author once the gate opens. Three choices were made while building it. The +> event is judged against the apply windows when it arrives, and a held create is noted for adoption +> only when it is released or a snapshot that holds it supersedes it. A snapshot supersedes only the +> events of its collection held before it was taken off the queue, and never a DELETE of an object +> it does not hold, since under the default prune mode a snapshot does not remove a document. The +> 2b fallback runs past a 10-minute hold or the branch's retained-byte budget: it drops what was +> held, counts it, and asks for a fresh snapshot. The sections below are the argument as it was +> made. The e2e spec `Audit route attribution` › *attributes a commit reached through a `rules[].sourceNamespace` override* has failed 9 times in CI since 2026-09-21, and passes on a diff --git a/docs/interpreting-metrics.md b/docs/interpreting-metrics.md index 71b6c7ae..3cd870ec 100644 --- a/docs/interpreting-metrics.md +++ b/docs/interpreting-metrics.md @@ -204,7 +204,7 @@ boundary, the commit, the push. Background: | `git_fetches_total` | counter | `provider_namespace`, `provider_name`, `branch`, `reason` | Every call that reads the remote through a `SmartFetch`. `reason` is `bootstrap` (repository preparation; **no production caller reaches it today**, so the series stays at zero) / `publication` (the head of a publication cycle) / `recovery` (re-establishing a base that could not be trusted, including a worktree a failed write left dirty that could not be undone locally) / `contention` (the reset onto the new tip after a push was rejected because somebody else moved the branch) / `push_failure_probe` (a push that failed without the remote saying anything, looking up where the branch is) / `forced_recheck` (a full re-read outside the publication cycle: a forced recheck, or the snapshot a resync judges against) / `refresh` (the periodic top-up of an idle branch, or a push receiver delivery on one, and only when the branch had actually moved). See the note below. | | `git_queue_refusals_total` | counter | `provider_namespace`, `provider_name`, `branch`, `kind` | Work refused at enqueue: the queue was full, intake was paused through an outage, or the worker was stopping. `kind` is `write` / `attach` / `resync` / `refresh`. The producer keeps a refused item and offers it again: the watch delivers a write again from its cursor, the controller re-sends a save, a resync is gathered again, and the next reconcile asks for a refresh. What it can cost, when its watch cursor expires before the write comes back, is the write's own commit and, under `prune.mode: OnEvent`, a deleted object's removal from Git. | | `git_commit_failures_total` | counter | `provider_namespace`, `provider_name`, `branch`, `kind`, `reason` | A window or request that died between routing and pushing. `kind` is `window` / `atomic`; `reason` is `refused` (a Git path a human must fix) / `error`. Every increment is a window's events lost until the next resync. An unreachable remote does not count: a decided window waits for the publication retry instead. | -| `git_writes_dropped_total` | counter | `provider_namespace`, `provider_name`, `branch`, `kind`, `reason` | A live event (`kind="event"`) or atomic write (`kind="atomic"`) the branch worker dropped because its `GitTarget`'s render fidelity was not established. `reason` is `render_fidelity_pending`: a collection of the `GitTarget` has not finished its replay, which closes every collection of that target until it does. Or it is `render_fidelity_diverged`: a rendered value does not match the live object. Nothing replays a dropped write, so each one is an edit missing from Git until the object changes again or its collection replays. The worker logs the first drop under each gate status at Info, naming the collections the gate waits for. | +| `git_writes_dropped_total` | counter | `provider_namespace`, `provider_name`, `branch`, `kind`, `reason` | A live event (`kind="event"`) or atomic write (`kind="atomic"`) the branch worker dropped because its `GitTarget`'s render fidelity was not established. `reason` is `render_fidelity_diverged`: a rendered value does not match the live object, and every write is dropped until a fresh replay measures the folder clean. Or it is `render_fidelity_pending`: a collection of the `GitTarget` has not finished its replay, which closes every collection of that target until it does. A live event that meets a pending gate is held, not dropped, and committed with its author once the gate opens; it is counted here only when the hold gives up, 10 minutes after the first one was held or once the branch holds more than its retained-byte budget, and the worker then asks for a fresh snapshot, which commits the current state as the operator (a deletion comes back only under prune mode `Always`). An atomic write is dropped at once. Nothing replays a dropped write, so each one is an edit missing from Git until the object changes again or its collection replays. The worker logs the first drop under each gate status at Info, naming the collections the gate waits for, and each hold, release and give-up at Info. | | `git_commit_windows_total` | counter | `gittarget_namespace`, `gittarget_name`, `close_reason`, `timer_source` | One per closed commit window, whether or not it then produced a commit. `close_reason` is `idle_timeout` / `max_duration` / `attach_next` / `identity_change` / `buffer_limit` / `resync_before_apply` / `atomic_before_apply` / `foreign_commit` (a Git host said the branch may have moved, so the window is published now) / `apply_due` (an apply came due for an applier on the branch, so the window's edits are published before the apply is sent) / `shutdown`; `timer_source` is `target` or `commit_request`. See [tuning commit windows](#tuning-commit-windows). | | `git_commit_window_duration_seconds` | histogram | `gittarget_namespace`, `gittarget_name` | How long each window collected: from the write that opened it to the start of its finalize. Its count equals `git_commit_windows_total` summed over `close_reason` and `timer_source`. | | `git_queue_depth` | gauge | `provider_namespace`, `provider_name`, `branch` | Pending + in-flight + committed-but-unpushed. Read at scrape time. | @@ -842,6 +842,7 @@ else pushed. See [who applies this folder](configuration.md#who-applies-this-fol | `apply_ownership_conflicts_total` | counter | `provider_namespace`, `provider_name`, `branch` | One per object another applier owns that the branch worker kept out of a coordinated folder: a live write not mirrored, or a snapshot object neither upserted nor swept. The owner is read from Flux's owner labels (`kustomize.toolkit.fluxcd.io/name` and `/namespace`) or Argo CD's tracking annotation (`argocd.argoproj.io/tracking-id`). Any non-zero rate is a misconfiguration: another `Kustomization` or `Application` applies objects into a namespace the folder mirrors, and two appliers applying one object fight each other. The worker logs each object once, naming both appliers; narrow the `WatchRule`, or move the object to one applier | | `stale_snapshots_total` | counter | `provider_namespace`, `provider_name`, `branch`, `outcome` | One per resync snapshot the branch worker found stale: collected while Git was ahead of the cluster (a commit pushed and not yet applied), collected across an apply, or arriving while Git is ahead. `outcome` is `discarded` (not decided; a fresh snapshot of its `GitTarget` is asked for once Git is no longer ahead, and the `GitTarget`'s live writes are held back by the render-fidelity gate until it lands) or `used` (the `GitTarget`'s snapshots had been held for 5 minutes while an apply kept failing, so this one was decided anyway, and whatever the unapplied commit changed in its objects is reverted, as the API-wins rule says). `discarded` comes in short bursts around applies and needs no action; a discard is not a resync failure and is not counted in `git_resync_failures_total`. Any `used` means an applier is failing for minutes at a time: read the applier claim's `status.operation` and the `ApplyFailed` Events. See [stale snapshots](spec/reconcile-via-watchlist-mark-and-sweep.md#stale-snapshots-are-discarded) | | `apply_operations_unresolved` | gauge | `provider_namespace`, `provider_name`, `branch` | Apply operations on the branch that have not completed 5 minutes after they were requested. Each is still followed, holds its applier's apply window open, and holds back the applier's next request until it completes, so a value that stays above zero is an applier stuck mid-apply: read its status, and the `ApplyUnresolved` Event on the `GitTarget`s. Read at scrape time | +| `appliers_held` | gauge | `provider_namespace`, `provider_name`, `branch` | Appliers on the branch with a reason pending that the initial-snapshot barrier holds back: no applier is asked for anything until every `GitTarget` naming its claim has had a snapshot of each collection it watches published, and its folder is on the branch. A short hold follows a start, a newly named claim or a watch change. A value that stays above zero is a `GitTarget` that is not mirrored: one whose `WatchRule` selects nothing holds its applier until it watches something, and one whose folder is missing waits for the fresh snapshot the worker asked for (a suspended `GitTarget` never writes it). The worker logs each hold, its reason and its end at Info ("An apply is held back"). Read at scrape time | | `receiver_deliveries_total` | counter | `receiver_namespace`, `receiver_name`, `type`, `outcome` | One per verified-or-refused delivery to a `GitReceiver`'s `/git-push/` path. `type` is its `spec.type`; `outcome` is `notified` (handed to at least one branch worker), `unknown_repository` (no `GitProvider`, in any namespace, has the repository the delivery names under any URL it carries), `unknown_branch` (one has, but no branch worker writes that branch for it), `ignored` (a ping, an event other than a push, or a tag or other non-branch ref), `rejected` (a missing or wrong signature or token, or an oversized body) or `invalid` (verified but unparseable, or naming no repository). Every outcome but `rejected` and `invalid` answers `202`. `unknown_repository` and `unknown_branch` are expected on an organization or GitHub App hook, which delivers for every repository and branch. On a repository's own hook, a steady `unknown_repository` rate means the payload names the repository by a URL no provider's `spec.url` matches, such as an SSH alias: the affected `GitBranch` has no `status.lastPushNoticeAt` | A delivery counted `notified` is only a hint: the worker reads the advertisement, and fetches under diff --git a/internal/applier/applier.go b/internal/applier/applier.go index 13bfeb52..58663d5b 100644 --- a/internal/applier/applier.go +++ b/internal/applier/applier.go @@ -226,6 +226,10 @@ type Started struct { // RequestedAt is when a Recovered operation was asked for, as its record says. A reconcile // that ran since is the only evidence of its apply this process can read. RequestedAt time.Time + // historyBefore is the highest status.history id an Argo CD Application held when this process + // set the sync's operation, -1 when it held none: an entry above it was recorded since. Nil for a + // Recovered operation, which no entry proves. + historyBefore *int64 // verdict is the outcome of a Flux apply, read once the resume's token was handled and kept // until the suspend after it is. verdict *Progress diff --git a/internal/applier/argocd.go b/internal/applier/argocd.go index ff10095e..580e47be 100644 --- a/internal/applier/argocd.go +++ b/internal/applier/argocd.go @@ -243,7 +243,8 @@ func (a *ArgoCD) setOperation( return Started{}, fmt.Errorf("build the sync operation for %s %s: %w", kindApplication, ref, err) } unstructured.RemoveNestedField(update.Object, "status", "operationState") - started := Started{Outcome: OutcomeSyncRequested, Token: token, Commit: ask.Commit} + before := lastHistoryID(app) + started := Started{Outcome: OutcomeSyncRequested, Token: token, Commit: ask.Commit, historyBefore: &before} _, err := a.client.Resource(applicationGVR()).Namespace(ref.Namespace). Update(ctx, update, metav1.UpdateOptions{}) switch { @@ -355,6 +356,9 @@ func (a *ArgoCD) Progress(ctx context.Context, ref types.NamespacedName, s *Star } return progress, nil } + if revision, synced := historySucceeded(app, s); synced { + return Progress{Outcome: git.ApplySucceeded, AppliedRevision: revision}, nil + } if s.Recovered || s.Unconfirmed != nil { return a.settleUnseen(ctx, ref, app, s) } @@ -392,6 +396,76 @@ func syncProgress(app *unstructured.Unstructured, token string) (Progress, bool) return Progress{}, false } +// historySucceeded reports the revision of the sync s asked for, from Argo CD's status.history, and +// false when no entry proves it ran. The controller appends an entry only for a sync that succeeded +// (appendHistory, controller/appcontroller.go), with its revision and initiatedBy, so the token can +// be gone and the sync still have succeeded: somebody synced again once ours completed, which +// replaced its operationState. Read as gone, it failed, and the worker asked again, so the commit +// was synced twice and its hooks ran twice. +// +// An entry proves it when it names the commit, GitOps Reverser as its initiator, and was recorded +// since the operation was set: above the highest id the Application held then. A Recovered operation +// never read the history, so no entry proves it. Its time cannot: an entry carries the controller's +// clock, to the second, and with that clock ahead an earlier sync of the commit reads as started +// after the request, which skipped an apply the request never made. Without the proof, a Recovered +// sync that ran and was then replaced is synced again, as before: that costs a repeat, never an apply. +func historySucceeded(app *unstructured.Unstructured, s *Started) (string, bool) { + if s.Commit == "" { + return "", false + } + history, _, _ := unstructured.NestedSlice(app.Object, "status", "history") + for _, item := range history { + entry, ok := item.(map[string]any) + if !ok { + continue + } + revision, _, _ := unstructured.NestedString(entry, "revision") + initiator, _, _ := unstructured.NestedString(entry, "initiatedBy", "username") + if !strings.EqualFold(revision, s.Commit) || initiator != ArgoCDInitiator || !s.recordedSince(entry) { + continue + } + return revision, true + } + return "", false +} + +// recordedSince reports whether a status.history entry was recorded after the operation was set +// (historySucceeded), which only an operation this process set can tell. +func (s *Started) recordedSince(entry map[string]any) bool { + if s.historyBefore == nil { + return false + } + id, ok := historyID(entry) + return ok && id > *s.historyBefore +} + +// lastHistoryID is the highest id in the Application's status.history, -1 when it holds none. +func lastHistoryID(app *unstructured.Unstructured) int64 { + last := int64(-1) + history, _, _ := unstructured.NestedSlice(app.Object, "status", "history") + for _, item := range history { + if entry, ok := item.(map[string]any); ok { + if id, ok := historyID(entry); ok && id > last { + last = id + } + } + } + return last +} + +// historyID is a status.history entry's id, which JSON decodes as int64 or float64. +func historyID(entry map[string]any) (int64, bool) { + switch id := entry["id"].(type) { + case int64: + return id, true + case float64: + return int64(id), true + case int: + return int64(id), true + } + return 0, false +} + // settleUnseen decides an operation whose update may have been sent without an answer (Unconfirmed, // or Recovered from a record written before the update), as app, read without its token, shows it. // The update carried the resourceVersion it was made from, so it can land only while the diff --git a/internal/applier/argocd_test.go b/internal/applier/argocd_test.go index 459d088c..f5db2a37 100644 --- a/internal/applier/argocd_test.go +++ b/internal/applier/argocd_test.go @@ -181,8 +181,11 @@ func TestArgoCDRequest_TheOperationIsWhatAutomatedSyncWouldBuild(t *testing.T) { }) require.NoError(t, err) - assert.Equal(t, Started{Outcome: OutcomeSyncRequested, Token: requestToken(stampTime), Commit: testCommit}, - started) + none := int64(-1) + assert.Equal(t, Started{ + Outcome: OutcomeSyncRequested, Token: requestToken(stampTime), Commit: testCommit, + historyBefore: &none, + }, started, "an Application with no history yet: every entry recorded is since") operation, found, _ := unstructured.NestedMap(readApp(t, client).Object, "operation") require.True(t, found) assert.Equal(t, map[string]any{ @@ -879,6 +882,124 @@ func TestArgoCD_Progress(t *testing.T) { assert.Equal(t, git.ApplyFailed, progress(t).Outcome, "ours was cleared before it recorded an outcome") } +// historyEntry is a status.history entry: a sync of revision that succeeded, started at startedAt. +func historyEntry(id int64, revision, initiator string, startedAt time.Time) map[string]any { + return map[string]any{ + "id": id, "revision": revision, "initiatedBy": map[string]any{"username": initiator}, + "deployStartedAt": startedAt.UTC().Format(time.RFC3339), + "deployedAt": startedAt.Add(2 * time.Second).UTC().Format(time.RFC3339), + } +} + +// TestArgoCD_AGoneTokenIsJudgedByTheHistory is A2: a sync somebody started once ours completed +// replaces its operationState, and the token with it. Read as gone, ours failed, so the worker +// synced the commit again and its hooks ran twice. Argo CD records every sync that succeeded in +// status.history, with its initiator: an entry of the commit, by GitOps Reverser, recorded since the +// operation was set, is ours succeeding. An entry from before it, or by somebody else, is not. +func TestArgoCD_AGoneTokenIsJudgedByTheHistory(t *testing.T) { + earlier := stampTime.Add(-time.Hour) + cases := []struct { + name string + history []any + want git.ApplyOutcome + }{ + { + name: "ours, then a manual sync", + history: []any{ + historyEntry(3, testCommit, ArgoCDInitiator, earlier), + historyEntry(4, testCommit, ArgoCDInitiator, stampTime), + historyEntry(5, olderCommit, "alice", stampTime.Add(time.Minute)), + }, + want: git.ApplySucceeded, + }, + { + name: "only an older sync of the same commit", + history: []any{historyEntry(3, testCommit, ArgoCDInitiator, earlier)}, + want: git.ApplyFailed, + }, + { + name: "a newer sync of the commit by somebody else", + history: []any{ + historyEntry(3, testCommit, ArgoCDInitiator, earlier), + historyEntry(4, testCommit, "alice", stampTime), + }, + want: git.ApplyFailed, + }, + { + name: "a newer sync by us of another commit", + history: []any{ + historyEntry(3, testCommit, ArgoCDInitiator, earlier), + historyEntry(4, olderCommit, ArgoCDInitiator, stampTime), + }, + want: git.ApplyFailed, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + app := manualApp() + _ = unstructured.SetNestedSlice(app.Object, []any{tc.history[0]}, "status", "history") + client := newArgoSource(app) + argo := pinnedArgo(client) + started, err := argo.Request(context.Background(), platformApp, Ask{ + Want: testBinding, Commit: testCommit, Reason: v1alpha3.ApplyReasonForeignCommit, + Token: requestToken(stampTime), + }) + require.NoError(t, err) + + app = readApp(t, client) + unstructured.RemoveNestedField(app.Object, "operation") + _ = unstructured.SetNestedMap(app.Object, map[string]any{ + "operation": theirOperation(), "phase": "Succeeded", "finishedAt": "2026-10-20T11:15:00Z", + }, "status", "operationState") + _ = unstructured.SetNestedSlice(app.Object, tc.history, "status", "history") + writeApp(t, client, app) + + got, err := argo.Progress(context.Background(), platformApp, &started) + require.NoError(t, err) + assert.Equal(t, tc.want, got.Outcome) + if tc.want == git.ApplySucceeded { + assert.Equal(t, testCommit, got.AppliedRevision) + } + }) + } +} + +// TestArgoCD_NoHistoryEntryProvesARecoveredSync: a Recovered operation never read the history, so +// only an entry's time could date it, and that is the controller's clock. With it ahead, an earlier +// sync of the commit read as started after the request, and a request that crashed before its update +// was Succeeded with the apply never made. No entry proves it now: the update is settled as never +// made (settleUnseen), and the worker asks again. +func TestArgoCD_NoHistoryEntryProvesARecoveredSync(t *testing.T) { + for _, tc := range []struct { + name string + started time.Time + }{ + {name: "an earlier sync, by a controller clock five seconds ahead", started: stampTime.Add(2 * time.Second)}, + {name: "started in the second after it was asked for", started: stampTime.Truncate(time.Second).Add(time.Second)}, + {name: "started in the second it was asked for", started: stampTime.Truncate(time.Second)}, + {name: "started a second before", started: stampTime.Add(-time.Second)}, + } { + t.Run(tc.name, func(t *testing.T) { + app := manualApp() + _ = unstructured.SetNestedSlice(app.Object, + []any{historyEntry(7, testCommit, ArgoCDInitiator, tc.started)}, "status", "history") + client := newArgoSource(app) + started := recoveredStarted(git.ApplyRequest{ + Applier: git.ApplierRef{Kind: v1alpha3.ApplierKindApplication}, + Recovered: &v1alpha3.ApplyOperation{ + Phase: v1alpha3.ApplyOperationRequested, Reason: v1alpha3.ApplyReasonRefusal, + Commit: testCommit, Token: requestToken(stampTime), RequestedAt: metav1.NewTime(stampTime), + }, + }) + + got, err := pinnedArgo(client).Progress(context.Background(), platformApp, &started) + + require.NoError(t, err) + assert.Equal(t, git.ApplyFailed, got.Outcome) + }) + } +} + // TestArgoCD_ASyncWhoseOperationWasRemovedEnds is review 7's A1. Removing .operation is the // community workaround for a stuck hook: Argo CD then processes the Application no further, and // its operationState stays Running. GitOps Reverser followed such a sync forever, and refused every diff --git a/internal/applier/flux.go b/internal/applier/flux.go index 215f5e84..abe641ee 100644 --- a/internal/applier/flux.go +++ b/internal/applier/flux.go @@ -141,7 +141,7 @@ func (f *Flux) fetchCommit( if artifactHolds(source, ask.Commit) { return nil } - source, err := f.fetch(ctx, sourceRef) + source, err := f.fetch(ctx, sourceRef, ask.Commit) if err != nil { return err } @@ -302,9 +302,16 @@ func (f *Flux) readSource( return source, nil } -// fetch stamps the GitRepository and waits for it to handle the stamp, which it marks once the -// fetch ended, whatever it found. A suspended Kustomization ignores the new artifact. -func (f *Flux) fetch(ctx context.Context, sourceRef types.NamespacedName) (*unstructured.Unstructured, error) { +// fetch stamps the GitRepository and waits for the fetch to end, which source-controller marks by +// handling the stamp, whatever it found. It also ends once the artifact holds commit, and once the +// GitRepository handled a stamp somebody wrote over ours: that fetch started after ours was asked +// for, so it read the remote no earlier than ours would have (tokenHandled). Waiting for our own +// token alone failed a request whose stamp `flux reconcile source git` or another controller +// replaced, after the whole wait, with the commit already in the artifact. A suspended +// Kustomization ignores the new artifact. +func (f *Flux) fetch( + ctx context.Context, sourceRef types.NamespacedName, commit string, +) (*unstructured.Unstructured, error) { token, err := f.stampSource(ctx, sourceRef) if err != nil { return nil, err @@ -315,7 +322,7 @@ func (f *Flux) fetch(ctx context.Context, sourceRef types.NamespacedName) (*unst if err != nil { return nil, fmt.Errorf("get %s %s: %w", kindGitRepository, sourceRef, err) } - if statusString(source, "lastHandledReconcileAt") == token { + if artifactHolds(source, commit) || tokenHandled(source, token) { return source, nil } select { @@ -566,8 +573,9 @@ func readyCondition(ks *unstructured.Unstructured) (metav1.Condition, bool) { return metav1.Condition{}, false } -// tokenHandled reports whether the reconcile that read token has ended: the Kustomization reports -// it handled, or reports handled a token stamped over it, which a later reconcile read. +// tokenHandled reports whether the reconcile that read token has ended: the object, a Kustomization +// or a GitRepository, reports it handled, or reports handled a token stamped over it, which a later +// reconcile read. func tokenHandled(ks *unstructured.Unstructured, token string) bool { handled := statusString(ks, "lastHandledReconcileAt") stamped := ks.GetAnnotations()[fluxmeta.ReconcileRequestAnnotation] diff --git a/internal/applier/flux_lifecycle_test.go b/internal/applier/flux_lifecycle_test.go index 7bcb4ceb..b7423534 100644 --- a/internal/applier/flux_lifecycle_test.go +++ b/internal/applier/flux_lifecycle_test.go @@ -472,6 +472,61 @@ func TestFluxRequest_AFetchThatEndsElsewhere(t *testing.T) { assert.True(t, suspended(sim.Kustomization(checkoutKey))) } +// stampOverOurs makes every stamp of the GitRepository platform land and be stamped over at once by +// another, as `flux reconcile source git` or another controller would: source-controller then +// handles that token, never ours. +func stampOverOurs(t *testing.T, sim *fluxsim.Sim) { + t.Helper() + sim.Client.PrependReactor("patch", "gitrepositories", func(k8stesting.Action) (bool, runtime.Object, error) { + obj, err := sim.Client.Tracker().Get(fluxsim.GitRepositoryGVR(), platformKey.Namespace, platformKey.Name) + if err != nil { + return true, nil, err + } + source, ok := obj.(*unstructured.Unstructured) + if !ok { + return true, nil, errors.New("not unstructured") + } + source = source.DeepCopy() + source.SetAnnotations(map[string]string{fluxmeta.ReconcileRequestAnnotation: "theirs"}) + return true, source, sim.Client.Tracker().Update(fluxsim.GitRepositoryGVR(), source, platformKey.Namespace) + }) +} + +// TestFluxRequest_AFetchStampedOverEndsWithTheirs is F2: a fetch whose stamp somebody else replaced +// waited for its own token, which source-controller never handles, and failed after the whole wait +// though the artifact held the commit. Now the fetch that read the stamp over ours ends ours too: +// the request goes on when it brought the commit, and fails at once, as any fetch elsewhere does, +// when it did not. +func TestFluxRequest_AFetchStampedOverEndsWithTheirs(t *testing.T) { + t.Run("the artifact holds the commit", func(t *testing.T) { + sim := simCluster(t, "main@sha1:"+olderCommit, true) + stampOverOurs(t, sim) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + flux := simFlux(sim) + + started, err := flux.Request(ctx, checkoutKey, ask(v1alpha3.ApplyReasonForeignCommit)) + + require.NoError(t, err, "no wait for a token nobody handles") + steps := follow(t, flux, &started) + assert.Equal(t, git.ApplySucceeded, steps[len(steps)-1].Outcome) + }) + t.Run("the artifact holds another commit", func(t *testing.T) { + sim := simCluster(t, "main@sha1:"+olderCommit, true) + sim.Fetches(platformKey, "main@sha1:"+olderCommit) + stampOverOurs(t, sim) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := simFlux(sim).Request(ctx, checkoutKey, ask(v1alpha3.ApplyReasonForeignCommit)) + + var elsewhere *sourceElsewhereError + require.ErrorAs(t, err, &elsewhere, "the fetch that read their stamp ended ours: %v", err) + require.NoError(t, ctx.Err(), "it did not wait the request out") + assert.True(t, suspended(sim.Kustomization(checkoutKey)), "nothing resumed") + }) +} + // TestFluxRequest_TakesSuspendBackFromAnUpdate: `flux resume` and `flux suspend` write spec.suspend // with a merge patch, once. That is no manifest owning the field, so a request takes it back. func TestFluxRequest_TakesSuspendBackFromAnUpdate(t *testing.T) { diff --git a/internal/controller/gittarget_controller.go b/internal/controller/gittarget_controller.go index 0eff8ae9..6c979b14 100644 --- a/internal/controller/gittarget_controller.go +++ b/internal/controller/gittarget_controller.go @@ -947,6 +947,10 @@ func (r *GitTargetReconciler) ensureEventStream( // It runs for every target on the branch and not only the one whose reconcile noticed. A worker // serves them all, and the siblings have no reason of their own to reconcile. // +// The record of which collections' snapshots settled goes the same way: the old worker published +// them, into the old repository, and the new worker's appliers must not be asked for anything until +// its own snapshots are in the new one. The worker holds them until this recovery is acknowledged. +// // The render-fidelity verdicts go the same way and for the same reason: they say a folder matches // live, about a folder in the other repository. They are INVALIDATED rather than forgotten — // forgetting a target leaves it unregistered, which the gate reads as writable, so the writes @@ -983,6 +987,7 @@ func (r *GitTargetReconciler) recoverBranchAfterReplacement( } ref := types.NewResourceReference(affected.Name, affected.Namespace).WithUID(string(affected.UID)) gate.Invalidate(ref) + r.EventRouter.WatchManager.RestartInitialSnapshots(ref) r.EventRouter.WatchManager.RequestRecheckForGitTarget(ref) recovered++ } diff --git a/internal/controller/gittarget_worker_repoint_test.go b/internal/controller/gittarget_worker_repoint_test.go index d99b9a2e..997d89f9 100644 --- a/internal/controller/gittarget_worker_repoint_test.go +++ b/internal/controller/gittarget_worker_repoint_test.go @@ -145,11 +145,23 @@ func TestRecoverBranchAfterReplacement_ReachesEverySiblingOnTheBranch(t *testing workers.RenderFidelityGate().Fail(appsRef, manifestanalyzer.RenderDivergence{}) require.False(t, workers.RenderFidelityGate().AllowsWrites(appsRef)) + // The snapshot records the old worker settled. + generations := map[string]uint64{} + for _, name := range []string{"apps", "infra", "release", "other"} { + generations[name] = manager.SnapshotGenerationForTest(types.NewResourceReference(name, "shop")) + } + branch := git.BranchKeyFor("https://example.invalid/second.git", "main") require.True(t, r.recoverBranchAfterReplacement(context.Background(), branch, shopRepo1Main, logr.Discard()), "the targets were listed and told, so the recovery is delivered") + for name, before := range generations { + restarted := manager.SnapshotGenerationForTest(types.NewResourceReference(name, "shop")) != before + assert.Equal(t, name == "apps" || name == "infra", restarted, + "%s: only the replaced worker's GitTargets start their snapshot record over", name) + } + assert.ElementsMatch(t, []string{"shop/apps", "shop/infra"}, manager.ForcedRecheckTargetsForTest(), "every GitTarget the replaced worker served, and nothing else") assert.Equal(t, git.RenderFidelityUnknown, workers.RenderFidelityGate().Status(appsRef).State, diff --git a/internal/git/applier_claim.go b/internal/git/applier_claim.go index 7c9714f6..7f4aac63 100644 --- a/internal/git/applier_claim.go +++ b/internal/git/applier_claim.go @@ -173,6 +173,8 @@ type branchTarget struct { // group names the GitRepository a Flux applier claim's Kustomization applies, in its cluster; empty // for an Argo CD Application or a source not read yet (apply_groups.go). group string + // folder is the GitTarget's spec.path, sanitized: the folder its applier is asked to apply. + folder string } // appliersOnBranch lists the GitTargets that write this worker's branch through its GitProvider and @@ -219,6 +221,7 @@ func (w *BranchWorker) appliersOnBranch(ctx context.Context) ([]branchTarget, er recorded: applierClaim.Operation(), repairPending: applierClaim.RepairPending(), group: sourceGroup(applierClaim), + folder: sanitizePath(target.Spec.Path), }) } sort.Slice(listed, func(i, j int) bool { return listed[i].ref.String() < listed[j].ref.String() }) @@ -257,10 +260,6 @@ type applierState struct { // finished it (apply_window.go). Nil when no attempt was sent, or the last asked for no apply // after the window before it had closed. window *applyWindow - // flushEdits records the API-side edits made since the flush mark of an attempt that is due and - // not sent yet, by objectKey: they are not in the commit the attempt applies, so its window holds - // them from the start (S1). Nil while no flush is waiting for this applier. - flushEdits map[string]time.Time // operation is the last operation an attempt started, as its applier claim's status.operation shows // it, and operationID the attempt it belongs to. Nil until this worker started one. operation *v1alpha3.ApplyOperation @@ -270,6 +269,9 @@ type applierState struct { // withdrawn is an operation an attempt recorded on the applier claim and then did not start, until // it is published as withdrawn. withdrawn *v1alpha3.ApplyOperation + // held is why the barrier holds back a reason pending for it, as last logged; empty while + // nothing is held (barrier.go). + held string // recovered is whether the applier claim's recorded status was read back for this applier, once, // when this process first listed it; followedSuspend is the suspend token of the last operation // the applier claim's reconciler recorded that this worker followed (apply_recovery.go). @@ -357,6 +359,7 @@ func (l *branchWorkerEventLoop) serviceAppliers() { return } defer l.publishClaimStatus() + defer l.publishHeld() now := time.Now() l.settleApplyResults(now) l.followRecordedOperations(now) @@ -377,7 +380,7 @@ func (l *branchWorkerEventLoop) serviceAppliers() { l.armApplierTimer(now, wait) return } - if !l.flushed(due) { + if !l.flushed() { // The publication wakes the loop when it lands (a push, or its retry); a save, a parent // recovery or a dirty checkout end with a wake of their own. l.armApplierTimer(now, wait) @@ -446,15 +449,24 @@ func (l *branchWorkerEventLoop) dueAppliers(now time.Time) ([]*applierState, tim l.w.Log.V(1).Info("The applier already applied the tip; the converge asks nothing", "branch", l.w.Branch, "applier", state.ref.String(), "commit", tip) } + held := "" switch { case !state.hasReason(refusals) || tip == "": - case !l.pastBarrier(state): - wait = earliest(wait, now.Add(barrierRecheckInterval)) + case l.barrierHold(state) != "": + held = l.barrierHold(state) case now.Before(state.retryAt): wait = earliest(wait, state.retryAt) default: - due = append(due, state) + // Looked at last: it reads the branch, and only an attempt that would go now needs it. + held = l.missingFolders(state) + if held == "" { + due = append(due, state) + } + } + if held != "" { + wait = earliest(wait, now.Add(barrierRecheckInterval)) } + l.noteHeld(state, held) } return due, wait } @@ -469,17 +481,11 @@ func (l *branchWorkerEventLoop) alreadyConverged( revisionNamesCommit(state.appliedRevision, tip) } -// pastBarrier reports whether every initial snapshot of every GitTarget the applier covers is -// published (initialSnapshotsPublished), so it may be asked for something. Before, an apply would -// destroy what is not mirrored yet: at startup the edits made while the operator was down, and for a -// GitTarget that newly names the applier claim, what its folder does not hold yet. +// pastBarrier reports whether the barrier lets the applier be asked for something (barrierHold), +// leaving out its folders, which only an attempt about to go reads (missingFolders). A folder already +// found missing still holds it. func (l *branchWorkerEventLoop) pastBarrier(state *applierState) bool { - for _, target := range state.targets { - if !l.initialSnapshotsPublished(target.ref) { - return false - } - } - return true + return l.barrierHold(state) == "" && !l.awaitsFolder(state) } // revisionNamesCommit reports whether an applier's revision names commit: Flux writes main@sha1:, @@ -502,20 +508,19 @@ func revisionCommit(revision string) string { // flushed makes the flush an apply needs (see serviceAppliers), and reports whether the writes up to // its mark are published, so the attempts may go. The first call of a flush closes the open window // and takes the mark; later calls only look whether the publication landed. -func (l *branchWorkerEventLoop) flushed(due []*applierState) bool { +// +// From the mark on, every edit is recorded for its GitTarget (flushEdits), not for the appliers due +// when the flush began: an applier that comes due while the flush waits for its publication is sent +// in the same flush, and an edit decided before it came due is not in the commit it applies either. +func (l *branchWorkerEventLoop) flushed() bool { if l.writeBusy() { return false } - for _, state := range due { - // A new mark publishes every edit recorded before it, so those are in the commit the - // attempt applies: carried into its window, they would withhold the apply's own write. - if state.flushEdits == nil || !l.flushing { - state.flushEdits = map[string]time.Time{} - } - } if !l.flushing { l.finalizeOpenWindowWithReason(windowFinalizeReasonApplyDue) - l.flushing, l.flushMark = true, l.decisions + // A new mark publishes every edit decided before it, so those are in the commit the attempts + // apply: carried into their windows, they would withhold the apply's own write. + l.flushing, l.flushMark, l.flushEdits = true, l.decisions, map[itypes.ResourceReference]map[string]time.Time{} if l.pendingUpTo(l.flushMark) { l.w.Log.Info("An apply is due; publishing the pending writes first", "branch", l.w.Branch, "pendingWrites", len(l.pendingWrites)) @@ -535,15 +540,42 @@ func (l *branchWorkerEventLoop) pendingUpTo(mark uint64) bool { return false } -// endFlush ends the flush once nothing is due: the edits recorded for an applier that has nothing -// pending any more are dropped with it. +// endFlush ends the flush once its attempts went, or nothing is due any more: the edits recorded +// since its mark are dropped with it, and the next flush records from its own mark. func (l *branchWorkerEventLoop) endFlush() { - l.flushing = false - for _, state := range l.appliers { - if state.inflight == nil && !state.hasReason(l.unsentRefusalsFor(state)) { - state.flushEdits = nil + l.flushing, l.flushEdits = false, nil +} + +// 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. +func (l *branchWorkerEventLoop) recordUnpublishedEdits(state *applierState) { + for _, target := range state.targets { + for key, at := range l.flushEdits[target.ref] { + state.window.recordEdit(key, at) } + if held := l.held[target.ref]; held != nil { + for _, h := range held.events { + state.window.recordEdit(objectKey(h.event.Identifier), h.event.LastFieldManagedAt) + } + } + } +} + +// recordFlushEdit records a live edit decided after the flush mark for its GitTarget, while a flush +// is waiting for its publication: it is not in the commit the flush's attempts apply. +func (l *branchWorkerEventLoop) recordFlushEdit(event Event) { + if !l.flushing { + return + } + target := itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace) + edits := l.flushEdits[target] + if edits == nil { + edits = map[string]time.Time{} + l.flushEdits[target] = edits } + key := objectKey(event.Identifier) + edits[key] = latest(edits[key], event.LastFieldManagedAt) } // earliest is the earlier of two deadlines, a zero one being none. @@ -579,10 +611,13 @@ func (l *branchWorkerEventLoop) syncAppliers(targets []branchTarget) { } named := map[ApplierRef][]branchTarget{} uids := map[itypes.ResourceReference]k8stypes.UID{} + listed := map[itypes.ResourceReference]bool{} for _, target := range targets { named[target.applier] = append(named[target.applier], target) uids[target.ref] = target.uid + listed[target.ref] = true } + l.forgetFolders(listed) for ref, state := range l.appliers { if _, ok := named[ref]; !ok { if state.inflight != nil { @@ -701,11 +736,7 @@ func (l *branchWorkerEventLoop) sendAttempt(state *applierState, tip string, ref state.lastTurn = attempt.id l.reportApplyWindow(state) state.openApplyWindow(attempt.id, tip, sentAt) - // The edits decided after the flush mark are not in the commit, and the apply may overwrite them. - for key, at := range state.flushEdits { - state.window.recordEdit(key, at) - } - state.flushEdits = nil + l.recordUnpublishedEdits(state) for key := range refusals { delete(l.unsentRefusals, key) } diff --git a/internal/git/applier_claim_test.go b/internal/git/applier_claim_test.go index 3bcc19d6..1a82e050 100644 --- a/internal/git/applier_claim_test.go +++ b/internal/git/applier_claim_test.go @@ -218,6 +218,25 @@ func TestServiceAppliers_NothingIsSentWhileNoFlushCanBeMade(t *testing.T) { } } +// TestServiceAppliers_AReplacedWorkerAsksNothingUntilItsGitTargetsWereTold is R4's worker half: a +// worker that replaced another starts while its GitTargets' snapshot record is still the old +// worker's, which says their snapshots settled. Until the recovery restarted that record, the +// startup converge would pass the barrier and ask for a folder the new repository does not hold. +func TestServiceAppliers_AReplacedWorkerAsksNothingUntilItsGitTargetsWereTold(t *testing.T) { + loop, requester := applierClaimLoop(t, applierTestTarget("alpha", true)) + pending := true + loop.w.replacementPending = func() bool { return pending } + observeForeignCommit(loop.w) + + loop.serviceAppliers() + assert.Empty(t, requester.taken(), "the GitTargets were not told yet") + assert.True(t, loop.appliers[applierRef("alpha")].foreign, "the reason waits") + + pending = false + loop.serviceAppliers() + assert.Len(t, requester.taken(), 1) +} + // argoApplier makes a GitTarget name the ArgoCDClaim of its own name, which holds the Application // argocd/. func argoApplier(g *configv1alpha3.GitTarget) { diff --git a/internal/git/applier_echoes_test.go b/internal/git/applier_echoes_test.go index 73c5db65..086dbf17 100644 --- a/internal/git/applier_echoes_test.go +++ b/internal/git/applier_echoes_test.go @@ -324,7 +324,7 @@ func TestApplierEchoes_FlushThenApply(t *testing.T) { assert.Nil(t, f.loop.openWindow, "the apply closed the open window") assert.Empty(t, requester.taken(), "the writes up to the mark are not published yet") - require.NotNil(t, f.echoApplier().flushEdits, "edits are recorded from the mark") + require.NotNil(t, f.loop.flushEdits, "edits are recorded from the mark") f.writeWindow(echoEvent("y", "y2")) // E1, decided after the mark require.NotNil(t, f.loop.openWindow, "it opens a window of its own") @@ -349,6 +349,98 @@ func TestApplierEchoes_FlushThenApply(t *testing.T) { assert.True(t, f.echoApplier().repair, "and a Repair puts E1 back once it is published") } +// TestApplierEchoes_AnApplierThatJoinsAFlushHoldsTheEditsSinceItsMark is S1 for an applier that comes +// due while a flush waits for its publication. It is sent in that flush, for the tip it produces, so +// an edit decided after the mark and before it came due is not in the commit it applies either. Its +// window recorded edits only from when it joined: the apply over such an edit was taken for the apply +// landing, not mirrored, and no Repair put the edit back on the cluster. +func TestApplierEchoes_AnApplierThatJoinsAFlushHoldsTheEditsSinceItsMark(t *testing.T) { + const late = "team-b" + f := newEchoFixture(t, "echo-flush-join", configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) + // Its folder is on the branch already, so the barrier holds nothing back for it. + createGitTargetWithPruneMode(t, f.worker, late, echoTarget, configv1alpha3.PruneOnEvent) + var lateTarget configv1alpha3.GitTarget + require.NoError(t, f.worker.Client.Get(f.worker.ctx, keyOf(late), &lateTarget)) + lateTarget.Spec.ApplierClaimRef = testApplier(late) + require.NoError(t, f.worker.Client.Update(f.worker.ctx, &lateTarget)) + createTestClaim(t, f.worker.Client, &lateTarget) + requester := f.withClaim() + require.Len(t, f.loop.appliers, 2) + applierFor := func(target string) *applierState { + for _, state := range f.loop.appliers { + if state.covers(Event{GitTargetName: target, GitTargetNamespace: "default"}) { + return state + } + } + t.Fatalf("no applier covers %s", target) + return nil + } + f.loop.lastPushAt = time.Now() + + f.writeWindow(echoEvent("x", "x2")) + restore := failPushes(t, f.worker) + applierFor(echoTarget).repair = true + f.loop.endWake(0) + require.Empty(t, requester.taken(), "the flush waits for its publication") + + edit := configMapTargetEvent("y", "alice", late) + edit.Operation = string(itypes.OperationUpdate) + f.writeWindow(edit) // decided after the mark, before team-b's applier comes due + applierFor(late).repair = true + f.loop.endWake(0) + require.Empty(t, requester.taken()) + + restore() + f.loop.retry.due = time.Now().Add(-time.Second) + f.loop.runRetry() + f.loop.endWake(0) + require.Len(t, requester.taken(), 2, "both appliers are sent in the one flush") + + applierFor(late).repair = false + over := configMapTargetEvent("y", "alice", late) + over.Operation = string(itypes.OperationUpdate) + over.LastFieldManagers = []string{FluxFieldManager} + over.LastFieldManagedAt = time.Now().Add(time.Hour) + assert.True(t, f.loop.withholdLateApply(over), "the apply over the edit is withheld") + assert.True(t, applierFor(late).repair, "and a Repair puts the edit back once it is published") +} + +// TestApplierEchoes_AFolderNotOnTheBranchIsCreatedBeforeItsApplierIsAsked is F-d: a GitTarget that +// names its claim after its snapshots settled empty has no folder, since only a resync of a GitTarget +// naming a claim creates one with no object. Its applier is not asked to apply a missing path: the +// barrier holds it, says so (F-a), and asks for one fresh snapshot, which is used even with a foreign +// commit pending, because the apply held for the folder would wait on it. +func TestApplierEchoes_AFolderNotOnTheBranchIsCreatedBeforeItsApplierIsAsked(t *testing.T) { + f := newEchoFixture(t, "echo-folder-missing", configv1alpha3.PruneOnEvent, nil) + var asked []itypes.ResourceReference + f.worker.resyncRecheck = func(target itypes.ResourceReference) { asked = append(asked, target) } + requester := f.withClaim() + state := f.echoApplier() + state.foreign = true + + f.loop.endWake(0) + f.loop.endWake(0) + assert.Empty(t, requester.taken(), "the folder is not on the branch") + assert.Equal(t, []itypes.ResourceReference{itypes.NewResourceReference(echoTarget, "default")}, asked, + "one fresh snapshot is asked for, once") + assert.Contains(t, state.held, "is not on the branch", "the hold says why") + assert.EqualValues(t, 1, f.worker.heldAppliers.Load()) + + req := &ResyncRequest{GitTargetName: echoTarget, GitTargetNamespace: "default", + Result: make(chan ResyncResult, 1)} + f.loop.handleResyncRequest(req) + f.loop.endWake(0) + require.NoError(t, (<-req.Result).Err, "used, not discarded for the pending foreign commit") + assert.NotEmpty(t, f.onMain(echoTarget+"/README.md"), "its resync created the folder") + + f.loop.endWake(0) + sent := requester.taken() + require.Len(t, sent, 1) + assert.Equal(t, configv1alpha3.ApplyReasonForeignCommit, sent[0].Reason) + assert.Empty(t, state.held, "the hold lifted") + assert.Zero(t, f.worker.heldAppliers.Load()) +} + // TestApplierEchoes_AResyncThatAddsAnObjectAdoptsIt is S2: an object a resync adds to Git runs on // the cluster already, and its applier never applied it, so a later Git-side delete would leave it // there. It is adopted; a resync that only updates adopts nothing. diff --git a/internal/git/apply_attempt_test.go b/internal/git/apply_attempt_test.go index 96d9f3df..4a6059ac 100644 --- a/internal/git/apply_attempt_test.go +++ b/internal/git/apply_attempt_test.go @@ -409,14 +409,20 @@ func TestApplyAttempt_ASourceElsewhereIsAHintToLook(t *testing.T) { func TestApplyAttempt_ANewFlushMarkStartsTheEditsOver(t *testing.T) { loop, _ := manualLoop(t, applierTestTarget("alpha", true)) loop.followRecordedOperations(time.Now()) - state := loop.appliers[applierRef("alpha")] - state.flushEdits = map[string]time.Time{"x": time.Now()} + alpha := itypes.NewResourceReference("alpha", "team-a") + edit := Event{ + GitTargetName: "alpha", GitTargetNamespace: "team-a", + Identifier: itypes.ResourceIdentifier{Resource: "configmaps", Namespace: "team-a", Name: "x"}, + } - loop.flushing = true - require.True(t, loop.flushed([]*applierState{state})) - assert.Len(t, state.flushEdits, 1, "the flush in progress keeps what it recorded") + require.True(t, loop.flushed()) + loop.recordFlushEdit(edit) + require.True(t, loop.flushed()) + assert.Len(t, loop.flushEdits[alpha], 1, "the flush in progress keeps what it recorded") - loop.flushing = false - require.True(t, loop.flushed([]*applierState{state})) - assert.Empty(t, state.flushEdits, "a new mark published them") + loop.endFlush() + loop.recordFlushEdit(edit) + assert.Empty(t, loop.flushEdits[alpha], "no flush, no record") + require.True(t, loop.flushed()) + assert.Empty(t, loop.flushEdits[alpha], "a new mark published them") } diff --git a/internal/git/apply_epoch.go b/internal/git/apply_epoch.go index ff8fabdf..406315d6 100644 --- a/internal/git/apply_epoch.go +++ b/internal/git/apply_epoch.go @@ -121,8 +121,13 @@ func (l *branchWorkerEventLoop) snapshotStale(req *ResyncRequest) string { // snapshotsWait reports whether F3 applies to a resync: the branch has an applier, the snapshot is // not its collection's initial one, and the GitTarget's snapshots have not been held for // snapshotHoldLimit already. A snapshot let through by the limit is logged and counted as used. +// +// The snapshot of a GitTarget whose folder the barrier waits for (missingFolders) is used too: the +// branch holds nothing of that folder for an apply to put on the cluster first, and the apply held +// for the folder would wait on the snapshot. func (l *branchWorkerEventLoop) snapshotsWait(req *ResyncRequest) bool { - if len(l.appliers) == 0 || req.Initial { + if len(l.appliers) == 0 || req.Initial || + l.awaitingFolders[itypes.NewResourceReference(req.GitTargetName, req.GitTargetNamespace)] { return false } since, held := l.snapshotHeldSince[resyncTargetKey(req)] diff --git a/internal/git/apply_window.go b/internal/git/apply_window.go index cbc95940..0cb4cc76 100644 --- a/internal/git/apply_window.go +++ b/internal/git/apply_window.go @@ -43,6 +43,17 @@ import ( // second-resolution times tie, with another manager or with the edit, the write could be either, and // it is mirrored as before; the collision is logged and counted as ambiguous so it is still seen. A // DELETE carries no managedFields of its deleter and is never withheld. +// +// Two clocks meet here. A send is the operator's time, a managedFields entry the source API server's, +// to the second. An entry is compared with an edit's, the same server's clock, wherever the window +// recorded an edit of the object, so a late apply over an edit and an undo of one are told apart +// whatever the skew. Only an object nobody edited in the window is judged against the send, and only +// to tell the apply landing from an undo of an edit published before it. A server clock behind the +// operator's dates the apply before the send: it is mirrored as an edit, which costs a commit of what +// Git holds, never an edit. A server clock ahead by more than the time between the earlier apply and +// the send dates that undo after the send: it is not mirrored, and the next apply reverts it. A skew +// under a second shifts neither across the send's second, and no grace is added: one would withhold +// such an undo when the earlier apply was that close to the send. // FluxFieldManager is the field manager kustomize-controller applies with (operation Apply). It is // fixed in Flux, not configurable (allFluxManagers in external-sources/flux/flux-operator @@ -213,9 +224,9 @@ func (state *applierState) covers(event Event) bool { // withholdLateApply reports whether a live event is the applier's own write in its apply window, // 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 a due -// attempt's window will hold once it is sent (applierState.flushEdits). It is called for each live -// event before it reaches the commit window. +// 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. func (l *branchWorkerEventLoop) withholdLateApply(event Event) bool { if len(l.appliers) == 0 { return false @@ -238,13 +249,10 @@ func (l *branchWorkerEventLoop) withholdLateApply(event Event) bool { l.recordApplyEcho(echoLate) } } - if applier.flushEdits != nil { - // Decided after the flush mark, so not in the commit the due attempt applies: its window - // holds it from the start (S1). - key := objectKey(event.Identifier) - applier.flushEdits[key] = latest(applier.flushEdits[key], event.LastFieldManagedAt) - } } + // 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) return false } @@ -269,9 +277,12 @@ func (l *branchWorkerEventLoop) judgeInWindow(applier *applierState, event Event l.recordApplyWindowCollision(applier, window, event, collisionAmbiguous) } window.recordEdit(key, written) - case written.Before(window.earliestSend(now).Truncate(time.Second)): + case !editedSince && written.Before(window.earliestSend(now).Truncate(time.Second)): // Older than every send still open: an earlier apply's entry, left newest by an update - // that removed every field the editor owned. It is an edit, the undo of one. + // that removed every field the editor owned. It is an edit, the undo of one. With an edit + // recorded, the undo is told by the edit's time instead (below), one server clock against + // itself: against the send, a server clock behind the operator's dated the apply over the + // edit before the send, and its revert was mirrored. window.recordEdit(key, written) case !editedSince: // The apply itself, on an object nobody edited since the attempt: its content is the @@ -289,8 +300,11 @@ 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. l.recordApplyWindowCollision(applier, window, event, collisionWithheld) - applier.repair = true + if !l.repairOnRelease(applier.ref, event) { + applier.repair = true + } return true } return false diff --git a/internal/git/apply_window_test.go b/internal/git/apply_window_test.go index ac98a896..42083a32 100644 --- a/internal/git/apply_window_test.go +++ b/internal/git/apply_window_test.go @@ -8,6 +8,7 @@ package git // apply_window.go. import ( + "fmt" "testing" "time" @@ -432,3 +433,68 @@ func TestApplyWindow_AnApplierWriteWithNoWindowIsLate(t *testing.T) { assert.False(t, loop.withholdLateApply(windowEvent("bravo", "2", FluxFieldManager)), "no applier covers bravo") assert.Equal(t, int64(1), echoes(t, reader, echoLate)) } + +// TestApplyWindow_ClockSkew is F-e: the send is the operator's clock and a managedFields time is the +// source API server's, to the second, so a skew between the two moves an entry across the send. +// Each case runs with the server's clock behind and ahead of the operator's by up to two seconds. +func TestApplyWindow_ClockSkew(t *testing.T) { + sent := time.Unix(1_800_000_000, 400_000_000) + type step struct { + at time.Duration // operator time of the write, from the send + managers []string + withheld bool + // echo is the apply landing on an object nobody edited: a server clock that dates it before + // the send makes it look like an earlier apply's entry left by an undo, and it is mirrored as + // an edit. That is the cost the window takes rather than ever withhold an undo. + echo bool + } + cases := []struct { + name string + steps []step + repair bool + }{ + { + name: "the apply lands on an object nobody edited", + steps: []step{ + {at: 300 * time.Millisecond, managers: []string{FluxFieldManager}, withheld: true, echo: true}, + }, + }, + { + name: "a late apply over a newer edit", + steps: []step{ + {at: 200 * time.Millisecond, managers: []string{"kubectl-edit"}}, + {at: 800 * time.Millisecond, managers: []string{FluxFieldManager}, withheld: true}, + }, + repair: true, + }, + { + name: "an undo of an edit made in the window leaves an earlier apply's entry", + steps: []step{ + {at: 200 * time.Millisecond, managers: []string{"kubectl-label"}}, + {at: -30 * time.Second, managers: []string{FluxFieldManager}}, + }, + }, + { + name: "an undo of an edit published before the send leaves an earlier apply's entry", + steps: []step{{at: -10 * time.Second, managers: []string{FluxFieldManager}}}, + }, + } + for _, skew := range []time.Duration{-2 * time.Second, -time.Second, 0, time.Second, 2 * time.Second} { + for _, tc := range cases { + t.Run(fmt.Sprintf("%s, server clock %+v", tc.name, skew), func(t *testing.T) { + loop, _ := sentAttempt(t) + state := loop.appliers[applierRef("alpha")] + state.window.sentAt = sent + for i, s := range tc.steps { + written := sent.Add(s.at + skew).Truncate(time.Second) + datedBeforeSend := written.Before(sent.Truncate(time.Second)) + want := s.withheld && (!s.echo || !datedBeforeSend) + assert.Equal(t, want, + loop.withholdLateApply(windowEventAt("alpha", "2", written, s.managers...)), + "step %d, %v", i, s.managers) + } + assert.Equal(t, tc.repair, state.repair, "repair") + }) + } + } +} diff --git a/internal/git/barrier.go b/internal/git/barrier.go new file mode 100644 index 00000000..163e56e1 --- /dev/null +++ b/internal/git/barrier.go @@ -0,0 +1,135 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "fmt" + + gogit "github.com/go-git/go-git/v6" + + itypes "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// This file is the barrier an apply waits for (S2 in docs/design/applier-echoes.md), and what it says +// while it holds one. No applier is asked for anything, for any reason, until: +// +// - the GitTargets of a worker that replaced another were told (WorkerManager.ReplacementPending): +// until then their record of settled snapshots is the old worker's; +// - every GitTarget the applier covers has had a snapshot of every collection it watches settled +// and published (initialSnapshotsPublished). One that watches no collection holds its applier, +// as a blocked watch does: nothing is mirrored into its folder yet; +// - the folder of each of those GitTargets is on the branch. A GitTarget that names an applier +// claim has its folder created by its resync even with no object, but one that named its claim +// after its snapshots settled empty has none, and Argo CD retries a sync of a missing path for +// minutes while Flux fails it. A fresh snapshot of it is asked for once, and its resync creates +// the folder. +// +// The hold is approved; the silence was the defect. Every applier the barrier holds a reason back +// for is logged at Info when the hold begins, changes reason or lifts, and counted in the +// appliers_held gauge. + +// barrierHold says why the barrier holds the applier back, leaving out its folders +// (missingFolders), or "" when it does not. +func (l *branchWorkerEventLoop) barrierHold(state *applierState) string { + if l.w.replacementPending != nil && l.w.replacementPending() { + return "this branch worker replaced one that wrote another repository, and the GitTargets on the " + + "branch have not been re-established yet" + } + for _, target := range state.targets { + if l.w.initialSnapshots != nil && !l.w.initialSnapshots(target.ref) { + return fmt.Sprintf("GitTarget %s has not had a snapshot of every collection it watches settled; one "+ + "that watches no collection holds its applier until it watches one", target.ref) + } + if !l.initialSnapshotsPublished(target.ref) { + return fmt.Sprintf("a snapshot of GitTarget %s is not published yet", target.ref) + } + } + return "" +} + +// awaitsFolder reports whether a GitTarget the applier covers is waiting for its folder +// (missingFolders). +func (l *branchWorkerEventLoop) awaitsFolder(state *applierState) bool { + for _, target := range state.targets { + if l.awaitingFolders[target.ref] { + return true + } + } + return false +} + +// missingFolders says which folder of a GitTarget the applier covers the branch does not hold, or "" +// when it holds every one. A missing folder's GitTarget is asked for a fresh snapshot once, and waits +// for it (awaitingFolders); its snapshot is then used whatever is ahead (snapshotsWait), since the +// apply held for it would wait on the snapshot. A branch that cannot be read is no reason to hold. +func (l *branchWorkerEventLoop) missingFolders(state *applierState) string { + var repo *gogit.Repository + missing := "" + for _, target := range state.targets { + if target.folder == "" { + continue + } + if repo == nil { + opened, err := gogit.PlainOpen(l.w.repoPath()) + if err != nil { + return "" + } + repo = opened + } + if headHasDir(repo, target.folder) { + delete(l.awaitingFolders, target.ref) + continue + } + if missing == "" { + missing = fmt.Sprintf("the folder %s of GitTarget %s is not on the branch; a fresh snapshot of the "+ + "GitTarget, whose resync creates it, was asked for", target.folder, target.ref) + } + if l.awaitingFolders[target.ref] { + continue + } + if l.awaitingFolders == nil { + l.awaitingFolders = map[itypes.ResourceReference]bool{} + } + l.awaitingFolders[target.ref] = true + if l.w.resyncRecheck != nil { + l.w.resyncRecheck(target.ref) + } + } + return missing +} + +// forgetFolders drops the folders awaited for GitTargets no longer on the branch. +func (l *branchWorkerEventLoop) forgetFolders(listed map[itypes.ResourceReference]bool) { + for ref := range l.awaitingFolders { + if !listed[ref] { + delete(l.awaitingFolders, ref) + } + } +} + +// noteHeld records why the barrier holds the applier back now, "" for nothing, and logs a change. +func (l *branchWorkerEventLoop) noteHeld(state *applierState, held string) { + if held == state.held { + return + } + if held == "" { + l.w.Log.Info("The barrier no longer holds the applier back", "branch", l.w.Branch, + "applier", state.ref.String(), "was", state.held) + } else { + l.w.Log.Info("An apply is held back: no applier is asked for anything before the GitTargets it covers "+ + "are mirrored", "branch", l.w.Branch, "applier", state.ref.String(), "reason", held) + } + state.held = held +} + +// publishHeld publishes how many of this branch's appliers the barrier holds a reason back for, for +// the appliers_held gauge. +func (l *branchWorkerEventLoop) publishHeld() { + var held int64 + for _, state := range l.appliers { + if state.held != "" { + held++ + } + } + l.w.heldAppliers.Store(held) +} diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index 33c48b0d..54cdc303 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -153,6 +153,9 @@ type BranchWorker struct { // unresolvedOperations is how many of this branch's apply operations are Unresolved, as the // loop last published it, for the apply_operations_unresolved gauge. unresolvedOperations atomic.Int64 + // heldAppliers is how many of this branch's appliers the barrier holds a reason back for, as the + // loop last published it, for the appliers_held gauge (barrier.go). + heldAppliers atomic.Int64 // applyEpoch, appliersAhead and hasAppliers are the loop's ApplyState for the watch side, which // reads it when a snapshot's collection starts (apply_epoch.go). applyEpoch atomic.Uint64 @@ -167,6 +170,10 @@ type BranchWorker struct { // admits every GitTarget. applierAdmitted ApplierAdmission resyncRecheck func(target itypes.ResourceReference) + // replacementPending reports whether this worker replaced one whose GitTargets have not been told + // yet (WorkerManager.ReplacementPending): until they are, their snapshot record is the old + // worker's, and no applier is asked for anything. Nil in the CLI and in tests. + replacementPending func() bool // Event processing eventQueue chan WorkItem @@ -304,7 +311,8 @@ type BranchWorker struct { retentionLoggedAt map[string]time.Time // hasUnpushedWork mirrors whether the event loop is currently holding a live - // open window or any committed-but-not-yet-pushed pending writes. The event + // open window, any committed-but-not-yet-pushed pending writes, or live events + // held at a pending render-fidelity gate (render_fidelity_hold.go). The event // loop is the only writer and, via syncUnpushedWorkFlag, the only reader; the // atomic guards the store/load and lets the depth gauge account for retained // work the channel length alone cannot see. @@ -1188,6 +1196,9 @@ type branchWorkerEventLoop struct { // apply_epoch.go. snapshotRerequests map[string]itypes.ResourceReference snapshotHeldSince map[string]time.Time + // awaitingFolders are the GitTargets whose folder the branch did not hold when their applier was + // about to be asked, and which a fresh snapshot was asked for, until it is there (barrier.go). + awaitingFolders map[itypes.ResourceReference]bool // ownershipLogged is the objects another applier owns already logged as kept out of a folder. // See apply_ownership.go. ownershipLogged map[string]struct{} @@ -1229,10 +1240,12 @@ type branchWorkerEventLoop struct { pushHinted bool // decisions numbers the decided writes (PendingWrite.decision). flushing is set while an apply is // due and waits for the writes up to flushMark, the last decision when it came due, to be - // published (serviceAppliers). - decisions uint64 - flushing bool - flushMark uint64 + // published (serviceAppliers). flushEdits holds the API-side edits decided since the mark, by + // GitTarget and objectKey: the attempts the flush sends hold them in their windows (S1). + decisions uint64 + flushing bool + flushMark uint64 + flushEdits map[itypes.ResourceReference]map[string]time.Time // 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: @@ -1246,6 +1259,14 @@ type branchWorkerEventLoop struct { // fidelityDropsLogged is, per GitTarget, the gate status under which its last drop was logged at // Info, so a closed gate logs its first drop and counts the rest. See render_fidelity_drops.go. fidelityDropsLogged map[string]renderFidelityDropMark + + // held is, per GitTarget, the live events held while its render fidelity is pending; heldSeq + // numbers them, heldEventBytes is what they are charged, and holdTimer looks at their gates again + // (render_fidelity_hold.go). + held map[itypes.ResourceReference]*heldWrites + heldSeq uint64 + heldEventBytes int64 + holdTimer *time.Timer } // commitWindowDefaults are a GitTarget's commit window timers when it declares none. @@ -1306,6 +1327,9 @@ func (l *branchWorkerEventLoop) run() { // serviceAppliers, at the end of the wake, sends what has come due. case <-l.w.applyWake: // An attempt reported back; serviceAppliers settles and sends. + case <-l.holdTimerC(): + l.holdTimer = nil + // serviceHeldWrites, at the start of the wake, looks at the held GitTargets' gates. } l.endWake(0) } @@ -1319,6 +1343,8 @@ func (l *branchWorkerEventLoop) handleQueueItem(item WorkItem) { // endWake is what every wake of the loop ends with, whatever woke it. released is the intake charge // of the item the wake handled, or zero. // +// - Release the live events held for a GitTarget whose render fidelity is now established, or drop +// them on a divergence or past the hold's bounds (serviceHeldWrites). // - Bind any waiting CommitRequest to an open window, resolve any whose attach deadline has // passed, and re-arm the deadline timer. // - Close the open window early if a Git host reported a push (closeWindowForPushHint). @@ -1329,6 +1355,7 @@ func (l *branchWorkerEventLoop) handleQueueItem(item WorkItem) { // ApplyState (apply_epoch.go). // - Publish the result, moving the handled item's charge to what the loop holds. func (l *branchWorkerEventLoop) endWake(released int64) { + l.serviceHeldWrites() l.serviceCommitRequests() l.closeWindowForPushHint() l.advance() @@ -1348,7 +1375,7 @@ func (l *branchWorkerEventLoop) endWake(released int64) { // publishing the flag second would let both read empty while a window is open, and a scrape landing // there would report a drained queue for a worker that has work. func (l *branchWorkerEventLoop) publishLoopState(released int64) { - l.w.hasUnpushedWork.Store(l.openWindow != nil || len(l.pendingWrites) > 0) + l.w.hasUnpushedWork.Store(l.openWindow != nil || len(l.pendingWrites) > 0 || len(l.held) > 0) l.syncAdmission(released) } @@ -1390,6 +1417,8 @@ func (l *branchWorkerEventLoop) dispatch(item WorkItem) int64 { // marker waited in the FIFO. Running it here keeps the original queue position, so a resync // still lands before the live events buffered behind it. req, charge := l.w.takePendingResync(item.Resync) + // Every live event held so far was queued before this snapshot was taken. + req.heldUpTo = l.heldSeq l.handleResyncRequest(req) return charge case item.Refresh != nil: @@ -1422,6 +1451,14 @@ func (l *branchWorkerEventLoop) handleLiveEvents(request *WriteRequest) { if !l.mirrors(event) { continue } + l.appendLiveEvent(event) + } +} + +// appendLiveEvent appends one live event that is mirrored to the commit window, splitting and +// finalizing it as identity, the byte cap or a zero window require. +func (l *branchWorkerEventLoop) appendLiveEvent(event Event) { + { if l.openWindow != nil && !l.openWindow.canAppend(event) { // Log the identity that broke the window so an unexpected split is // diagnosable: the common cause is an incoming event whose author is empty @@ -1471,7 +1508,7 @@ func (l *branchWorkerEventLoop) handleLiveEvents(request *WriteRequest) { l.noteForeignWindow() l.attachWaitingCommitRequests() if l.openWindow == nil { - continue // the attached request's maxDuration: 0s closed it already + return // the attached request's maxDuration: 0s closed it already } } @@ -1482,28 +1519,40 @@ func (l *branchWorkerEventLoop) handleLiveEvents(request *WriteRequest) { // fails on a stuck remote will not free memory, but that's the // known stuck-push pathology documented in the design. l.finalizeOpenWindowWithReason(windowFinalizeReasonBufferLimit) - continue + return } l.closeOrArmWindow() } } -// mirrors reports whether a live event reaches the commit window. It does not while the target's -// render fidelity is not established, for an object another applier owns in a folder an applier claim -// applies (F4), and for the applier's own write in its apply window (apply_window.go). +// mirrors reports whether a live event reaches the commit window now. It does not for an object +// another applier owns in a folder an applier claim applies (F4), for the applier's own write in its +// apply window (apply_window.go), and while the target's render fidelity is not established: on a +// 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 { - if status, closed := l.w.renderFidelityClosed(event.GitTargetName, event.GitTargetNamespace); closed { - l.dropAtRenderFidelity(itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace), - status, writeDroppedEvent, event.Identifier.String()) + target := itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace) + l.serviceHeld(target) + status, closed := l.w.renderFidelityClosed(event.GitTargetName, event.GitTargetNamespace) + if closed && status.State == RenderFidelityFalse { + l.dropAtRenderFidelity(target, status, writeDroppedEvent, event.Identifier.String()) return false } if l.ownedElsewhere(event.GitTargetName, event.GitTargetNamespace, event.Identifier.String(), event.AppliedBy) { return false } + // Judged when it arrives, held or not: the apply windows and the flush record an edit when it + // was made. 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) + return false + } if event.Operation == string(itypes.OperationCreate) { l.noteCreated(event) } @@ -1546,6 +1595,7 @@ func (l *branchWorkerEventLoop) handleAtomicRequest(request *WriteRequest) { func (l *branchWorkerEventLoop) handleShutdown() { l.w.Log.Info("Handling shutdown, finalizing open window and pushing pending commits") l.finalizeOpenWindowWithReason(windowFinalizeReasonShutdown) + l.dropHeldAtShutdown() if len(l.pendingWrites) > 0 { // Shutdown bypasses the cooldown and the retry: pending work needs to land before the // worker exits, even if a push was just sent. The push commits what was just decided. @@ -1892,6 +1942,7 @@ func (l *branchWorkerEventLoop) stopTimers() { l.stopAttachTimer() l.stopRefusalTimer() l.stopApplierTimer() + l.stopHoldTimer() if l.retry.timer != nil { l.retry.timer.Stop() l.retry.timer = nil diff --git a/internal/git/intake.go b/internal/git/intake.go index 7dfa8645..7d3147e0 100644 --- a/internal/git/intake.go +++ b/internal/git/intake.go @@ -169,10 +169,11 @@ func (r *ResyncRequest) payloadSize() int64 { func (r *ResyncRequest) charge() int64 { return r.payloadSize() + pendingWriteOverheadBytes } // heldBytes is what the loop holds that is not published yet: the open window, the pending writes, -// and the saves waiting for a window. An attached save is in the charge of the window it is attached -// to, and then of the write that window became. +// the live events held at a pending render-fidelity gate, and the saves waiting for a window. An +// attached save is in the charge of the window it is attached to, and then of the write that window +// became. func (l *branchWorkerEventLoop) heldBytes() int64 { - held := l.windowBytes + l.pendingWritesBytes + held := l.windowBytes + l.pendingWritesBytes + l.heldEventBytes for _, pcr := range l.pendingCRs { if !pcr.attached { held += attachCharge(pcr.message) diff --git a/internal/git/pending_writes_loop.go b/internal/git/pending_writes_loop.go index 22861bd6..fe7d6023 100644 --- a/internal/git/pending_writes_loop.go +++ b/internal/git/pending_writes_loop.go @@ -369,6 +369,8 @@ func (l *branchWorkerEventLoop) settleResyncApplied(i int) { // caller once it has the answer: the next item off the queue may be a live event, and it must // meet the gate this resync produced. See recordReplayClean. l.recordReplayClean(req) + // Its snapshot holds what the live events of its collection held before it did. + l.supersedeHeld(req) l.w.Log.Info("Resync request applied", "committed", committed, "created", stats.Created, diff --git a/internal/git/publication.go b/internal/git/publication.go index 2c253075..b28a431e 100644 --- a/internal/git/publication.go +++ b/internal/git/publication.go @@ -154,6 +154,9 @@ func (m *WorkerManager) setPublicationGaugeSources() { telemetry.SetGaugeSource(telemetry.GaugeApplyOperationsUnresolved, source(func(w *BranchWorker) (int64, bool) { return always(w.unresolvedOperations.Load()) })) + telemetry.SetGaugeSource(telemetry.GaugeAppliersHeld, source(func(w *BranchWorker) (int64, bool) { + return always(w.heldAppliers.Load()) + })) } // publicationGauges are the gauges setPublicationGaugeSources installs. @@ -166,6 +169,7 @@ var publicationGauges = []string{ telemetry.GaugeGitOldestRetainedWrite, telemetry.GaugeGitNextRetry, telemetry.GaugeApplyOperationsUnresolved, + telemetry.GaugeAppliersHeld, } // publicationEventBuffer bounds the GitTarget events a burst of report changes can queue. A full diff --git a/internal/git/publication_test.go b/internal/git/publication_test.go index ceede003..7f3e0766 100644 --- a/internal/git/publication_test.go +++ b/internal/git/publication_test.go @@ -113,6 +113,10 @@ func TestPublication_MetricsDescribeTheBacklog(t *testing.T) { unresolved, ok := gauge(telemetry.GaugeApplyOperationsUnresolved) require.True(t, ok) assert.Equal(t, int64(1), unresolved, "an operation past its budget is still followed, and seen") + f.worker.heldAppliers.Store(1) + held, ok := gauge(telemetry.GaugeAppliersHeld) + require.True(t, ok) + assert.Equal(t, int64(1), held, "an applier the barrier holds back is seen") f.worker.branchBufferMaxBytes = 1 loop.publishLoopState(0) diff --git a/internal/git/render_fidelity_drops.go b/internal/git/render_fidelity_drops.go index 58506db6..5199f3b3 100644 --- a/internal/git/render_fidelity_drops.go +++ b/internal/git/render_fidelity_drops.go @@ -91,6 +91,11 @@ func (l *branchWorkerEventLoop) dropAtRenderFidelity( } else { log.V(1).Info("Dropped a write because the GitTarget's render fidelity is not established") } + l.recordWriteDropped(kind, reason) +} + +// recordWriteDropped counts one write dropped at the render-fidelity gate. +func (l *branchWorkerEventLoop) recordWriteDropped(kind, reason string) { if telemetry.GitWritesDroppedTotal == nil { return } diff --git a/internal/git/render_fidelity_drops_test.go b/internal/git/render_fidelity_drops_test.go index 09eafaf3..4e6132b9 100644 --- a/internal/git/render_fidelity_drops_test.go +++ b/internal/git/render_fidelity_drops_test.go @@ -141,9 +141,10 @@ func TestRenderFidelity_AReplayFromARetiredStreamDoesNotOpenTheGate(t *testing.T assert.False(t, f.gate.AllowsWrites(f.target), "only the current stream's replay opens the gate") } -// TestRenderFidelity_ADroppedWriteIsCountedAndLoggedOncePerGateStatus: a write the closed gate -// drops is gone for good, so every drop is counted, and the first under each gate status is -// logged at Info with what the gate waits for. A burst of edits against one closed gate logs once. +// TestRenderFidelity_ADroppedWriteIsCountedAndLoggedOncePerGateStatus: a pending gate holds live +// writes (render_fidelity_hold.go), and a divergence drops them, with every later one. A write the +// gate drops is gone for good, so every drop is counted, and the first under each gate status is +// logged at Info with what the gate waits for or found. A burst against one closed gate logs once. func TestRenderFidelity_ADroppedWriteIsCountedAndLoggedOncePerGateStatus(t *testing.T) { reader, err := telemetry.InitTestExporter() require.NoError(t, err) @@ -158,19 +159,23 @@ func TestRenderFidelity_ADroppedWriteIsCountedAndLoggedOncePerGateStatus(t *test }}) assert.Nil(t, f.loop.openWindow, "a pending scope closes the GitTarget") - assert.Equal(t, int64(2), writesDropped(t, reader, writeDroppedEvent, writeDropRenderFidelityPending)) + require.NotNil(t, f.loop.held[f.target]) + assert.Len(t, f.loop.held[f.target].events, 2, "its live writes are held, not dropped") + assert.Len(t, logged.matching("Holding the GitTarget's live writes"), 1, "the hold is said once") + assert.Zero(t, writesDropped(t, reader, writeDroppedEvent, writeDropRenderFidelityPending)) assert.Equal(t, int64(1), writesDropped(t, reader, writeDroppedAtomic, writeDropRenderFidelityPending)) drops := logged.matching("Dropped a write") - require.Len(t, drops, 1, "one Info line for the burst") + require.Len(t, drops, 1) assert.Contains(t, drops[0], "Waiting for 1 of 1", "it says which scopes the gate waits for") - assert.Contains(t, drops[0], "configmaps/first") _, applied := f.gate.RecordScopeDivergence(f.target, f.revision, f.scope.Collection, manifestanalyzer.RenderDivergence{Field: "data.region", Token: "${REGION}"}) require.True(t, applied) f.liveEvent("third") - assert.Equal(t, int64(1), writesDropped(t, reader, writeDroppedEvent, writeDropRenderFidelityDiverged)) - assert.Len(t, logged.matching("Dropped a write"), 2, "a new gate status logs again") + assert.Nil(t, f.loop.held[f.target], "a divergence drops what was held") + assert.Equal(t, int64(3), writesDropped(t, reader, writeDroppedEvent, writeDropRenderFidelityDiverged)) + assert.Len(t, logged.matching("Dropped a write"), 2, "a new gate status logs again, once") + assert.Nil(t, f.loop.openWindow) } // infoLines collects the worker's Info-level log lines. diff --git a/internal/git/render_fidelity_hold.go b/internal/git/render_fidelity_hold.go new file mode 100644 index 00000000..69afd272 --- /dev/null +++ b/internal/git/render_fidelity_hold.go @@ -0,0 +1,316 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +import ( + "slices" + "sort" + "time" + + itypes "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// This file holds a GitTarget's live events while its render fidelity is pending, rather than drop +// them (Fix 2 in docs/design/live-edit-dropped-while-render-fidelity-pending.md). +// +// The gate is per GitTarget and its scopes per collection, so one collection that is still +// replaying (a new or restarted collection, every collection at operator start, a snapshot F3 +// discarded) closes live writes for all of them. A dropped event never comes back: the watch has +// moved past it. So an event that meets a pending gate is judged at once, as on arrival (the +// ownership check, the apply windows), and then HELD: +// +// - When the gate turns True, the GitTarget's held events are released in arrival order, before +// any later event of it, and committed with their authors. A create is noted for adoption then, +// not on arrival: an apply sent before the create is in a commit stamps nothing, and spends the +// adopt (F-c, which lost it outright: the gate returned before the create was noted at all). +// - When it turns False, a divergence, they are dropped and counted, as before: that is the gate +// doing its job. +// - An applied snapshot of a collection supersedes the events of that collection held before it +// was taken off the queue: they are older than its list, which holds their effect. A create it +// supersedes is noted for adoption, since the snapshot's commit holds the object. A DELETE of an +// object the snapshot does not hold stays held, because a snapshot removes a document only under +// prune mode Always. +// - An apply sent while events are held applies a commit without them, and may overwrite them on +// the cluster. Its window holds them as edits from the send (recordUnpublishedEdits), so the +// applier's write over one is withheld. Its Repair waits for the event to reach Git, by release +// or a snapshot superseding it (repairOnRelease): sent before, it applies a commit without the +// edit and is spent, and a Repair pending reads as Git ahead, so F3 discards the very snapshot +// the gate waits for. +// - The hold is bounded, per GitTarget by renderFidelityHoldLimit, and per branch by the +// retained-byte budget. Past either, the GitTarget's held events are dropped and counted, and a +// fresh snapshot of it is asked for, which brings the state back as the operator's commit. A +// DELETE among them comes back only under prune mode Always; the log says how many there were. +// +// Held events count against the intake budget like the open window and the pending writes. + +// renderFidelityHoldLimit is how long a GitTarget's live events may be held, from the first one: at +// least as long as F3 holds its snapshots (snapshotHoldLimit), so a discarded snapshot's fresh one +// supersedes them rather than the limit. +const renderFidelityHoldLimit = 2 * snapshotHoldLimit + +// renderFidelityRecheckInterval is how soon a GitTarget whose events are held looks at its gate +// again: the gate changes on other goroutines (the watch layer's resync drain, a plan change) without +// waking the loop. +const renderFidelityRecheckInterval = time.Second + +// heldWrites is one GitTarget's live events held while its render fidelity is pending. +type heldWrites struct { + // since is when the first of them was held. + since time.Time + events []heldEvent + // deletes counts the DELETEs among them. + deletes int +} + +// heldEvent is one live event held, numbered in the order the loop took it. overwrittenBy is the +// appliers whose apply wrote over it while it was held: each has a Repair due once it reaches Git. +type heldEvent struct { + seq uint64 + event Event + charge int64 + overwrittenBy []ApplierRef +} + +// holdAtRenderFidelity holds a live event of a GitTarget whose render fidelity is pending. +func (l *branchWorkerEventLoop) holdAtRenderFidelity(target itypes.ResourceReference, event Event, + status RenderFidelityStatus, +) { + held := l.held[target] + if held == nil { + if l.held == nil { + l.held = map[itypes.ResourceReference]*heldWrites{} + } + held = &heldWrites{since: l.w.now()} + l.held[target] = held + l.w.Log.Info("Holding the GitTarget's live writes until its render fidelity is established; they "+ + "are committed in order once it is", "gitTarget", target.String(), "renderFidelity", status.Message) + } + l.heldSeq++ + charge := eventRetainedCharge(&event) + held.events = append(held.events, heldEvent{seq: l.heldSeq, event: event, charge: charge}) + if event.Operation == string(itypes.OperationDelete) { + held.deletes++ + } + l.heldEventBytes += charge + l.w.Log.V(1).Info("Held a live write while the GitTarget's render fidelity is pending", + "gitTarget", target.String(), "resource", event.Identifier.String(), "held", len(held.events)) + if budget := l.w.branchBufferMaxBytes; budget > 0 && l.heldEventBytes > budget { + for _, ref := range l.heldTargets() { + l.giveUpHold(ref, "the live writes held for the branch exceed its retained-byte budget") + } + } +} + +// serviceHeldWrites settles every GitTarget whose events are held, at the start of each wake, and +// arms the recheck while any are. +func (l *branchWorkerEventLoop) serviceHeldWrites() { + for _, target := range l.heldTargets() { + l.serviceHeld(target) + } + l.stopHoldTimer() + if len(l.held) > 0 { + l.holdTimer = time.NewTimer(renderFidelityRecheckInterval) + } +} + +// serviceHeld settles one GitTarget's held events by its gate now: released once it is True, +// dropped once it is False, given up once the hold reached its limit. +func (l *branchWorkerEventLoop) serviceHeld(target itypes.ResourceReference) { + held := l.held[target] + if held == nil { + return + } + status, closed := l.w.renderFidelityClosed(target.Name, target.Namespace) + switch { + case !closed: + l.releaseHeld(target, held) + case status.State == RenderFidelityFalse: + l.takeHeld(target) + for _, h := range held.events { + l.dropAtRenderFidelity(target, status, writeDroppedEvent, h.event.Identifier.String()) + } + case l.w.now().Sub(held.since) >= renderFidelityHoldLimit: + l.giveUpHold(target, "the GitTarget's render fidelity has been pending for "+ + renderFidelityHoldLimit.String()) + } +} + +// releaseHeld commits a GitTarget's held events, in the order they were held. +func (l *branchWorkerEventLoop) releaseHeld(target itypes.ResourceReference, held *heldWrites) { + l.takeHeld(target) + l.w.Log.Info("The GitTarget's render fidelity is established; committing the live writes held for it", + "gitTarget", target.String(), "writes", len(held.events), + "heldFor", l.w.now().Sub(held.since).Round(time.Millisecond).String()) + for _, h := range held.events { + if h.event.Operation == string(itypes.OperationCreate) { + l.noteCreated(h.event) + } + l.appendLiveEvent(h.event) + l.repairOverwritten(h) + } +} + +// giveUpHold drops a GitTarget's held events past the hold's bounds, and asks for a fresh snapshot of +// it, which re-derives what they changed. +func (l *branchWorkerEventLoop) giveUpHold(target itypes.ResourceReference, why string) { + held := l.takeHeld(target) + if held == nil { + return + } + for range held.events { + l.recordWriteDropped(writeDroppedEvent, writeDropRenderFidelityPending) + } + l.w.Log.Info("Dropped the live writes held for a GitTarget, and asked for a fresh snapshot of it, which "+ + "commits the current state as the operator. A deletion comes back only under prune mode Always", + "gitTarget", target.String(), "reason", why, "writes", len(held.events), "deletions", held.deletes) + if l.w.resyncRecheck != nil { + l.w.resyncRecheck(target) + } +} + +// supersedeHeld drops the events of a GitTarget's collection held before a snapshot of that +// collection was taken off the queue (req.heldUpTo), once the snapshot is applied: it is newer than +// each of them, and holds their effect. A DELETE of an object the snapshot does not hold is kept, +// since only prune mode Always removes a document the snapshot lacks. Events of other collections, +// and every event held since, stay held. +func (l *branchWorkerEventLoop) supersedeHeld(req *ResyncRequest) { + target := itypes.NewResourceReference(req.GitTargetName, req.GitTargetNamespace) + held := l.held[target] + if held == nil { + return + } + inSnapshot := make(map[string]bool, len(req.Desired)) + for _, dr := range req.Desired { + inSnapshot[objectKey(dr.Resource)] = true + } + kept := held.events[:0] + superseded := 0 + for _, h := range held.events { + if !supersededBy(req, h, inSnapshot) { + kept = append(kept, h) + continue + } + switch itypes.OperationType(h.event.Operation) { + case itypes.OperationDelete: + held.deletes-- + case itypes.OperationCreate: + l.noteCreated(h.event) + case itypes.OperationUpdate: + } + // The snapshot may have been listed before the apply wrote over the event, and so hold it. + l.repairOverwritten(h) + l.heldEventBytes -= h.charge + superseded++ + } + held.events = kept + if superseded == 0 { + return + } + l.w.Log.V(1).Info("A snapshot superseded live writes held for its collection", + "gitTarget", target.String(), "scope", req.Scope.String(), "superseded", superseded, + "stillHeld", len(kept)) + if len(kept) == 0 { + delete(l.held, target) + } +} + +// supersededBy reports whether an applied resync's snapshot supersedes a held event: the event was +// held before the snapshot was taken off the queue, is of its collection, and is not a DELETE of an +// object the snapshot does not hold (inSnapshot, by objectKey). +func supersededBy(req *ResyncRequest, h heldEvent, inSnapshot map[string]bool) bool { + if h.seq > req.heldUpTo || !snapshotCovers(req, h.event) { + return false + } + return h.event.Operation != string(itypes.OperationDelete) || inSnapshot[objectKey(h.event.Identifier)] +} + +// snapshotCovers reports whether a resync's snapshot is of the collection an event came from: the +// stream's collection, selector included, when the event names it, and its boundary otherwise. +func snapshotCovers(req *ResyncRequest, event Event) bool { + if req.Scope == nil { + return true + } + if event.SourceCollection != (itypes.CollectionKey{}) { + return event.SourceCollection == req.Scope.Collection + } + return req.Scope.Matches(event.Identifier) +} + +// repairOnRelease reports whether a live event the applier wrote over is held, and if so leaves a +// Repair due for the applier once the latest held event of that object reaches Git +// (repairOverwritten). The commit an apply sent now applies does not hold the edit yet. +func (l *branchWorkerEventLoop) repairOnRelease(ref ApplierRef, event Event) bool { + held := l.held[itypes.NewResourceReference(event.GitTargetName, event.GitTargetNamespace)] + if held == nil { + return false + } + key := objectKey(event.Identifier) + for i := len(held.events) - 1; i >= 0; i-- { + h := &held.events[i] + if objectKey(h.event.Identifier) != key { + continue + } + if !slices.Contains(h.overwrittenBy, ref) { + h.overwrittenBy = append(h.overwrittenBy, ref) + } + return true + } + return false +} + +// repairOverwritten leaves a Repair pending for each applier that wrote over a held event, now that +// the event reaches Git. An applier no longer listed has nothing to repair. +func (l *branchWorkerEventLoop) repairOverwritten(h heldEvent) { + for _, ref := range h.overwrittenBy { + if state := l.appliers[ref]; state != nil { + state.repair = true + } + } +} + +// takeHeld removes a GitTarget's held events from the hold, and returns them. +func (l *branchWorkerEventLoop) takeHeld(target itypes.ResourceReference) *heldWrites { + held := l.held[target] + if held == nil { + return nil + } + delete(l.held, target) + for _, h := range held.events { + l.heldEventBytes -= h.charge + } + return held +} + +// heldTargets is the GitTargets whose events are held, in a stable order. +func (l *branchWorkerEventLoop) heldTargets() []itypes.ResourceReference { + targets := make([]itypes.ResourceReference, 0, len(l.held)) + for target := range l.held { + targets = append(targets, target) + } + sort.Slice(targets, func(i, j int) bool { return targets[i].String() < targets[j].String() }) + return targets +} + +// dropHeldAtShutdown says what an exiting loop leaves held: nothing replays it. +func (l *branchWorkerEventLoop) dropHeldAtShutdown() { + for _, target := range l.heldTargets() { + held := l.takeHeld(target) + l.w.Log.Info("Shutting down with live writes held for a GitTarget whose render fidelity is pending; "+ + "they are missing from Git until the objects change again or the collections replay", + "gitTarget", target.String(), "writes", len(held.events)) + } +} + +func (l *branchWorkerEventLoop) holdTimerC() <-chan time.Time { + if l.holdTimer == nil { + return nil + } + return l.holdTimer.C +} + +func (l *branchWorkerEventLoop) stopHoldTimer() { + if l.holdTimer != nil { + l.holdTimer.Stop() + l.holdTimer = nil + } +} diff --git a/internal/git/render_fidelity_hold_test.go b/internal/git/render_fidelity_hold_test.go new file mode 100644 index 00000000..8f169fa4 --- /dev/null +++ b/internal/git/render_fidelity_hold_test.go @@ -0,0 +1,363 @@ +// SPDX-License-Identifier: Apache-2.0 + +package git + +// Live events held while a GitTarget's render fidelity is pending (render_fidelity_hold.go): Window +// B of docs/design/live-edit-dropped-while-render-fidelity-pending.md, which Fix 1 cannot reach. + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime/schema" + + configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/manifestanalyzer" + "github.com/ConfigButler/gitops-reverser/internal/telemetry" + itypes "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// heldEvent is a live event of the fixture's ConfigMaps collection, by author. +func (f *replayFixture) streamEvent(name, author string, operation itypes.OperationType) Event { + event := configMapTargetEvent(name, author, ledgerTargetName) + event.Operation = string(operation) + event.SourceCollection = f.scope.Collection + if operation == itypes.OperationDelete { + event.Object = nil + } + return event +} + +func (f *replayFixture) send(events ...Event) { + f.loop.handleQueueItem(WorkItem{Request: &WriteRequest{Events: events, CommitMode: CommitModePerEvent}}) +} + +// windowNames is the objects the open window holds, in the order they were added. +func (f *replayFixture) windowNames() []string { + if f.loop.openWindow == nil { + return nil + } + var names []string + for _, event := range f.loop.openWindow.orderedEvents() { + names = append(names, event.Identifier.Name) + } + return names +} + +// TestRenderFidelityHold_AnEditMadeWhileAnotherCollectionReplaysIsCommittedByItsAuthor is Window B: +// an edit that arrives while some collection of the GitTarget is still replaying met a pending gate +// and was dropped for good. It is held, and committed with its author once the gate opens, though +// nothing wakes the loop but its recheck: the watch layer's drain records the scope on its own +// goroutine. +func TestRenderFidelityHold_AnEditMadeWhileAnotherCollectionReplaysIsCommittedByItsAuthor(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + f := newReplayFixture(t, "render-fidelity-hold-window-b") + + f.send(f.streamEvent("edited", "alice", itypes.OperationUpdate)) + require.Nil(t, f.loop.openWindow) + require.NotNil(t, f.loop.holdTimer, "the held GitTarget's gate is looked at again") + assert.Positive(t, f.loop.heldBytes(), "held writes count against the intake budget") + + _, applied := f.gate.RecordScopeClean(f.target, f.revision, f.scope.Collection) + require.True(t, applied) + f.loop.endWake(0) // the recheck + + assert.Equal(t, []string{"edited"}, f.windowNames(), "released once the gate opened") + assert.Equal(t, "alice", f.loop.openWindow.Author, "with its author") + assert.Empty(t, f.loop.held) + assert.Zero(t, f.loop.heldEventBytes) + assert.Zero(t, writesDropped(t, reader, writeDroppedEvent, writeDropRenderFidelityPending)) +} + +// TestRenderFidelityHold_HeldEditsGoBeforeLaterOnes: a gate can open between two wakes, and the next +// event of the GitTarget must not overtake the ones held for it. +func TestRenderFidelityHold_HeldEditsGoBeforeLaterOnes(t *testing.T) { + f := newReplayFixture(t, "render-fidelity-hold-order") + f.send(f.streamEvent("first", "alice", itypes.OperationUpdate)) + _, applied := f.gate.RecordScopeClean(f.target, f.revision, f.scope.Collection) + require.True(t, applied) + + f.send(f.streamEvent("second", "bob", itypes.OperationUpdate)) + + require.NotNil(t, f.loop.openWindow) + assert.Equal(t, "bob", f.loop.openWindow.Author, + "alice's held edit went first, and bob's closed her window") + assert.Equal(t, []string{"second"}, f.windowNames()) +} + +// TestRenderFidelityHold_ASnapshotSupersedesTheEditsOfItsCollectionHeldBeforeIt: the replay resync +// of a collection is newer than the events of it held before it was taken off the queue, and holds +// their effect, so applying them after it would write older state over newer. They are dropped. A +// DELETE of an object the snapshot does not hold is kept: a snapshot removes a document only under +// prune mode Always. Events of another collection, and events held after the snapshot was taken, +// stay held. +func TestRenderFidelityHold_ASnapshotSupersedesTheEditsOfItsCollectionHeldBeforeIt(t *testing.T) { + f := newReplayFixture(t, "render-fidelity-hold-supersede") + secrets := itypes.CollectionKeyFor(schema.GroupVersionResource{Version: "v1", Resource: "secrets"}, "default") + _, revisions := f.gate.Reconcile(f.target, + []itypes.CollectionKey{f.scope.Collection, secrets}, []itypes.CollectionKey{secrets}) + require.Equal(t, f.revision, revisions[f.scope.Collection], "precondition: configmaps kept its stream") + + other := configMapTargetEvent("other-collection", "alice", ledgerTargetName) + other.SourceCollection = secrets + f.send( + f.streamEvent("keep", "alice", itypes.OperationUpdate), // in the snapshot: superseded + f.streamEvent("gone", "alice", itypes.OperationDelete), // not in it: kept + f.streamEvent("recreated", "alice", itypes.OperationDelete), // in it again: superseded + other, + ) + require.Len(t, f.loop.held[f.target].events, 4) + + req := f.replay() + req.Desired = append(req.Desired, desiredCMIn("recreated", "default", "green")) + f.loop.handleQueueItem(WorkItem{Resync: &resyncMarker{request: req}}) + require.NoError(t, (<-req.Result).Err) + f.send(f.streamEvent("after", "alice", itypes.OperationUpdate)) // held after the snapshot: kept + + held := f.loop.held[f.target] + require.NotNil(t, held, "secrets is still pending") + var names []string + for _, h := range held.events { + names = append(names, h.event.Identifier.Name) + } + assert.Equal(t, []string{"gone", "other-collection", "after"}, names) + assert.Equal(t, 1, held.deletes) +} + +// TestRenderFidelityHold_ACreateHeldIsAdoptedOnceItIsInAWrite is F-c: an adopt was lost while render +// fidelity was closed, because the gate returned before the create was noted. Noted on arrival, it +// would be sent before the create is in any commit, which stamps nothing and spends it. A held create +// is noted when it is released, or when a snapshot that holds it supersedes it, so the CatchUp goes +// after a flush that publishes it. +func TestRenderFidelityHold_ACreateHeldIsAdoptedOnceItIsInAWrite(t *testing.T) { + setup := func(t *testing.T, slug string) (*echoFixture, *recordingRequester, *RenderFidelityGate, uint64) { + t.Helper() + f := newEchoFixture(t, slug, configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) + requester := f.withClaim() + gate := NewRenderFidelityGate() + f.worker.renderFidelityGate = gate + _, revisions := gate.Reconcile(itypes.NewResourceReference(echoTarget, "default"), + []itypes.CollectionKey{echoScope().Collection}, []itypes.CollectionKey{echoScope().Collection}) + created := configMapTargetEvent("created", "alice", echoTarget) + created.SourceCollection = echoScope().Collection + f.writeWindow(created) + require.Nil(t, f.loop.openWindow, "held") + require.Empty(t, requester.taken(), "nothing to adopt yet: the create is in no commit") + return f, requester, gate, revisions[echoScope().Collection] + } + + t.Run("released", func(t *testing.T) { + f, requester, gate, revision := setup(t, "render-fidelity-hold-adopt-released") + gate.RecordScopeClean(itypes.NewResourceReference(echoTarget, "default"), revision, echoScope().Collection) + f.loop.endWake(0) + + sent := requester.taken() + require.Len(t, sent, 1) + assert.Equal(t, configv1alpha3.ApplyReasonCatchUp, sent[0].Reason) + assert.NotEmpty(t, f.onMain(echoFile("created")), "the flush published the create first") + }) + + t.Run("superseded", func(t *testing.T) { + f, requester, _, revision := setup(t, "render-fidelity-hold-adopt-superseded") + scope := echoScope() + // The snapshot holds the object as Git already does: it adds nothing, so only the + // superseded create says it was the API's. + gitIn(t, f.clone, "pull", "--quiet", "--ff-only", "origin", "main") + f.pushFromOutside("created", "created") + req := &ResyncRequest{GitTargetName: echoTarget, GitTargetNamespace: "default", Scope: &scope, + RenderFidelityRevision: revision, Initial: true, Result: make(chan ResyncResult, 1), + Desired: []manifestanalyzer.DesiredResource{echoDesired("x", "x1"), echoDesired("created", "created")}} + f.loop.handleQueueItem(WorkItem{Resync: &resyncMarker{request: req}}) + require.NoError(t, (<-req.Result).Err) + + require.Len(t, requester.taken(), 1, "the push from outside is a foreign commit, which is asked") + require.NotNil(t, f.echoApplier().inflight) + assert.True(t, f.echoApplier().inflight.adopt, "and the attempt carries the adopt") + assert.Empty(t, f.loop.held) + }) +} + +// TestRenderFidelityHold_AnApplyOverAHeldEditRepairsItOnceItIsInGit: an apply sent while an edit is +// held applies a commit without it. Its window recorded only the flush's edits, so the applier's +// write over the held edit was taken for the apply landing: the edit then reached Git with the +// cluster still running the overwritten value, and no Repair. The window now holds the held events +// from the send, and the Repair waits for the edit to reach Git: sent before, it applies a commit +// without the edit and is spent, and pending, it reads as Git ahead, so F3 discards the snapshot the +// gate waits for. +func TestRenderFidelityHold_AnApplyOverAHeldEditRepairsItOnceItIsInGit(t *testing.T) { + target := itypes.NewResourceReference(echoTarget, "default") + configMaps := echoScope().Collection + secrets := itypes.CollectionKeyFor(schema.GroupVersionResource{Version: "v1", Resource: "secrets"}, "default") + setup := func(t *testing.T, slug string) (*echoFixture, *recordingRequester, map[itypes.CollectionKey]uint64) { + t.Helper() + f := newEchoFixture(t, slug, configv1alpha3.PruneOnEvent, map[string]string{"x": "x1"}) + requester := f.withClaim() + f.worker.initialSnapshots = func(itypes.ResourceReference) bool { return true } + gate := NewRenderFidelityGate() + f.worker.renderFidelityGate = gate + // ConfigMaps replayed clean; Secrets is replaying again, so the GitTarget's events are held. + _, revisions := gate.Reconcile(target, []itypes.CollectionKey{configMaps, secrets}, + []itypes.CollectionKey{configMaps, secrets}) + gate.RecordScopeClean(target, revisions[configMaps], configMaps) + edit := echoEvent("x", "x2") + edit.SourceCollection = configMaps + edit.LastFieldManagers = []string{"kubectl-edit"} + edit.LastFieldManagedAt = time.Now() + f.writeWindow(edit) + require.Len(t, f.loop.held[target].events, 1) + + // A push from outside is applied while the edit is held. + f.echoApplier().foreign = true + f.loop.endWake(0) + sent := requester.taken() + require.Len(t, sent, 1) + require.Equal(t, echoDocument("x", "x1"), f.onMain(echoFile("x")), "the commit sent does not hold the edit") + + over := echoEvent("x", "x1") + over.SourceCollection = configMaps + over.LastFieldManagers = []string{FluxFieldManager} + over.LastFieldManagedAt = time.Now().Add(time.Second) + assert.True(t, f.loop.withholdLateApply(over), "the apply over the held edit is withheld") + report(sent[0], ApplySucceeded) + f.loop.endWake(0) + assert.Empty(t, requester.taken(), "no Repair goes before the edit is in Git") + assert.False(t, f.loop.appliersAhead(), "nor is one pending, which would hold back the Secrets snapshot") + return f, requester, revisions + } + repaired := func(t *testing.T, f *echoFixture, requester *recordingRequester) { + t.Helper() + f.loop.finalizeOpenWindowWithReason(windowFinalizeReasonIdleTimeout) + f.loop.pushPending() + f.loop.endWake(0) + assert.Equal(t, echoDocument("x", "x2"), f.onMain(echoFile("x"))) + sent := requester.taken() + require.Len(t, sent, 1, "the cluster runs x1, so the commit that holds the edit is applied") + assert.Equal(t, configv1alpha3.ApplyReasonRepair, sent[0].Reason) + assert.Equal(t, gitIn(t, f.repoDir, "rev-parse", "main"), sent[0].Commit) + } + + // The Secrets replay is not the collection's first, so F3 would discard its snapshot with a Repair + // pending, and ask again only once that Repair went. + replaySecrets := func(t *testing.T, f *echoFixture, revisions map[itypes.CollectionKey]uint64) { + t.Helper() + scope := ResyncScope{Collection: secrets} + req := &ResyncRequest{GitTargetName: echoTarget, GitTargetNamespace: "default", Scope: &scope, + RenderFidelityRevision: revisions[secrets], Result: make(chan ResyncResult, 1)} + f.loop.handleQueueItem(WorkItem{Resync: &resyncMarker{request: req}}) + require.NoError(t, (<-req.Result).Err, "the snapshot is used, not discarded") + } + + t.Run("released", func(t *testing.T) { + f, requester, revisions := setup(t, "render-fidelity-hold-overwritten-released") + replaySecrets(t, f, revisions) + f.loop.endWake(0) + repaired(t, f, requester) + }) + + t.Run("across a second apply", func(t *testing.T) { + f, requester, revisions := setup(t, "render-fidelity-hold-overwritten-twice") + f.echoApplier().foreign = true + f.loop.endWake(0) + sent := requester.taken() + require.Len(t, sent, 1, "another apply goes while the edit is held") + over := echoEvent("x", "x1") + over.SourceCollection = configMaps + over.LastFieldManagers = []string{FluxFieldManager} + over.LastFieldManagedAt = time.Now().Add(2 * time.Second) + assert.True(t, f.loop.withholdLateApply(over), "its window holds the edit from the send") + report(sent[0], ApplySucceeded) + f.loop.endWake(0) + require.Empty(t, requester.taken()) + require.Len(t, f.loop.held[target].events, 1, "the edit is still held") + + replaySecrets(t, f, revisions) + f.loop.endWake(0) + repaired(t, f, requester) + }) + + t.Run("superseded", func(t *testing.T) { + f, requester, revisions := setup(t, "render-fidelity-hold-overwritten-superseded") + // A snapshot of ConfigMaps listed before the apply wrote over the edit holds it. + scope := echoScope() + req := &ResyncRequest{GitTargetName: echoTarget, GitTargetNamespace: "default", Scope: &scope, + RenderFidelityRevision: revisions[configMaps], Initial: true, Result: make(chan ResyncResult, 1), + Desired: []manifestanalyzer.DesiredResource{echoDesired("x", "x2")}} + f.loop.handleQueueItem(WorkItem{Resync: &resyncMarker{request: req}}) + require.NoError(t, (<-req.Result).Err) + require.Empty(t, f.loop.held, "the snapshot superseded the edit") + repaired(t, f, requester) + }) +} + +// echoScope is the echo fixture's ConfigMaps collection. +func echoScope() ResyncScope { + return ResyncScopeFor(schema.GroupVersionResource{Version: "v1", Resource: "configmaps"}, "default") +} + +// TestRenderFidelityHold_AHoldPastItsLimitAsksForAFreshSnapshot: a scope can stay pending for a long +// time, so the hold is bounded. Past the limit the GitTarget's held writes are dropped and counted, +// and a fresh snapshot of it is asked for, which commits the current state as the operator. +func TestRenderFidelityHold_AHoldPastItsLimitAsksForAFreshSnapshot(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + f := newReplayFixture(t, "render-fidelity-hold-limit") + now := time.Now() + f.worker.clock = func() time.Time { return now } + var asked []itypes.ResourceReference + f.worker.resyncRecheck = func(target itypes.ResourceReference) { asked = append(asked, target) } + logged := captureInfo(f.worker) + + f.send(f.streamEvent("edited", "alice", itypes.OperationUpdate), + f.streamEvent("removed", "alice", itypes.OperationDelete)) + now = now.Add(renderFidelityHoldLimit - time.Second) + f.loop.endWake(0) + require.NotNil(t, f.loop.held[f.target], "not yet") + + now = now.Add(time.Second) + f.loop.endWake(0) + assert.Empty(t, f.loop.held, "given up") + assert.Zero(t, f.loop.heldEventBytes) + assert.Equal(t, []itypes.ResourceReference{f.target}, asked, "a fresh snapshot is asked for") + assert.Equal(t, int64(2), writesDropped(t, reader, writeDroppedEvent, writeDropRenderFidelityPending)) + gaveUp := logged.matching("Dropped the live writes held") + require.Len(t, gaveUp, 1) + assert.Contains(t, gaveUp[0], `"deletions"=1`, "it says how many deletions a snapshot may not bring back") +} + +// TestRenderFidelityHold_TheBranchBudgetBoundsTheHold: held writes are charged like a window, and +// past the branch's retained-byte budget they are given up at once rather than grow. +func TestRenderFidelityHold_TheBranchBudgetBoundsTheHold(t *testing.T) { + f := newReplayFixture(t, "render-fidelity-hold-budget") + var asked []itypes.ResourceReference + f.worker.resyncRecheck = func(target itypes.ResourceReference) { asked = append(asked, target) } + event := f.streamEvent("edited", "alice", itypes.OperationUpdate) + f.worker.branchBufferMaxBytes = eventRetainedCharge(&event) + 1 + + f.send(event) + require.NotNil(t, f.loop.held[f.target]) + f.send(f.streamEvent("again", "alice", itypes.OperationUpdate)) + + assert.Empty(t, f.loop.held) + assert.Equal(t, []itypes.ResourceReference{f.target}, asked) +} + +// TestRenderFidelityHold_ADivergenceAfterAReplayDropsWhatWasHeld: a divergence is the gate doing its +// job, so what was held for the GitTarget is dropped, as a write arriving then would be. +func TestRenderFidelityHold_ADivergenceAfterAReplayDropsWhatWasHeld(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + f := newReplayFixture(t, "render-fidelity-hold-diverged") + f.send(f.streamEvent("edited", "alice", itypes.OperationUpdate)) + + _, applied := f.gate.RecordScopeDivergence(f.target, f.revision, f.scope.Collection, + manifestanalyzer.RenderDivergence{Field: "data.region", Token: "${REGION}"}) + require.True(t, applied) + f.loop.endWake(0) + + assert.Empty(t, f.loop.held) + assert.Nil(t, f.loop.openWindow) + assert.Equal(t, int64(1), writesDropped(t, reader, writeDroppedEvent, writeDropRenderFidelityDiverged)) +} diff --git a/internal/git/types.go b/internal/git/types.go index 1bbce359..d70ad18e 100644 --- a/internal/git/types.go +++ b/internal/git/types.go @@ -595,6 +595,10 @@ type ResyncRequest struct { // payloadBytes is the snapshot's estimated serialized size once sized is set. See payloadSize. payloadBytes int64 sized bool + // heldUpTo is the last live event the loop had held at a pending render-fidelity gate when it + // took this request off the queue: those were queued before the snapshot was taken, so once it + // applies it supersedes the ones of its collection (render_fidelity_hold.go). + heldUpTo uint64 } // refusalCollection is the watched collection this request speaks for: its scope's collection for a per-type @@ -778,8 +782,9 @@ type Event struct { BootstrapOptions pathBootstrapOptions // SourceCollection names the target-watch collection that produced this event. Zero for every - // non-stream producer (reconcile, bootstrap, the admission path). Diagnostic only: - // nothing filters the queue on it. See source_collection.go. + // non-stream producer (reconcile, bootstrap, the admission path). Nothing filters the queue on + // it; a snapshot of that collection supersedes the event while it is held at a pending + // render-fidelity gate (render_fidelity_hold.go). See source_collection.go. SourceCollection types.CollectionKey // payloadBytes is the object's estimated serialized size once sized is set. See payloadSize. diff --git a/internal/git/worker_manager.go b/internal/git/worker_manager.go index e6af9909..5b5d1cdb 100644 --- a/internal/git/worker_manager.go +++ b/internal/git/worker_manager.go @@ -513,6 +513,7 @@ func (m *WorkerManager) startWorkerLocked(key BranchKey, req WorkerRequest) erro worker.claimStatus = m.claimStatus worker.initialSnapshots = m.initialSnapshots worker.resyncRecheck = m.resyncRecheck + worker.replacementPending = func() bool { return m.ReplacementPending(key) } worker.applierAdmitted = m.applierAdmitted worker.renderFidelityGate = m.renderFidelityGate worker.crOwners = m.commitRequests diff --git a/internal/git/write_gate.go b/internal/git/write_gate.go index bda05261..f849c7f6 100644 --- a/internal/git/write_gate.go +++ b/internal/git/write_gate.go @@ -36,8 +36,8 @@ var ( // errTargetSuspended is a request's failure on a suspended GitTarget, which writes nothing. errTargetSuspended = errors.New("the GitTarget is suspended") // errRenderFidelityClosed is a request's failure while the GitTarget's render-vs-live check has - // not passed. The gate drops the target's writes on arrival, so the save cannot know what it - // would have contained. + // not passed. The gate holds the target's live writes until it opens, or drops them on a + // divergence, so the save cannot know in time what it would have contained. errRenderFidelityClosed = errors.New( "the GitTarget's render fidelity is not established, so its writes are held back") ) diff --git a/internal/telemetry/exporter.go b/internal/telemetry/exporter.go index dcb0a58b..18351c63 100644 --- a/internal/telemetry/exporter.go +++ b/internal/telemetry/exporter.go @@ -745,6 +745,7 @@ func registerObservableGauges() error { {"gitopsreverser_resource_condition", GaugeResourceCondition}, {"gitopsreverser_git_branch_targets", GaugeGitBranchTargets}, {"gitopsreverser_apply_operations_unresolved", GaugeApplyOperationsUnresolved}, + {"gitopsreverser_appliers_held", GaugeAppliersHeld}, } for _, o := range observable { if _, err := otelMeter.Int64ObservableGauge( diff --git a/internal/telemetry/gauges.go b/internal/telemetry/gauges.go index 7c63ad67..c2feb0d7 100644 --- a/internal/telemetry/gauges.go +++ b/internal/telemetry/gauges.go @@ -64,6 +64,10 @@ const ( // within the dispatcher's budget and are still followed, each holding its applier's apply // window open and its next attempt back. Read from what the branch worker's loop last published. GaugeApplyOperationsUnresolved = "apply_operations_unresolved" + // GaugeAppliersHeld is how many appliers on a branch have a reason pending that the initial + // snapshot barrier holds back: a GitTarget they cover is not mirrored yet, or its folder is not on + // the branch. Read from what the branch worker's loop last published. + GaugeAppliersHeld = "appliers_held" ) var ( diff --git a/internal/watch/apply_state.go b/internal/watch/apply_state.go index 2c4be6b8..4ed8d4ef 100644 --- a/internal/watch/apply_state.go +++ b/internal/watch/apply_state.go @@ -44,27 +44,59 @@ func (m *Manager) InitialSnapshotsSettled(gitDest types.ResourceReference) bool return len(collections) > 0 && m.snapshotsSettled(gitDest, collections) } -// markSnapshotSettled records that a snapshot of the collection settled: a later one of it is no -// longer initial (git.ResyncRequest.Initial). -func (m *Manager) markSnapshotSettled(gitDest types.ResourceReference, collection types.CollectionKey) { +// snapshotRecord is one GitTarget's settled snapshots, under the generation they were settled for. +type snapshotRecord struct { + generation uint64 + settled map[types.CollectionKey]bool +} + +// snapshotGeneration is the generation a snapshot of the GitTarget enqueued now settles under +// (markSnapshotSettled). A record is started for a GitTarget that has none. +func (m *Manager) snapshotGeneration(gitDest types.ResourceReference) uint64 { m.snapshotsMu.Lock() defer m.snapshotsMu.Unlock() + return m.snapshotRecordLocked(gitDest).generation +} + +// snapshotRecordLocked is the GitTarget's record, started under a fresh generation when it has none. +// Called with snapshotsMu held. +func (m *Manager) snapshotRecordLocked(gitDest types.ResourceReference) *snapshotRecord { if m.settledSnapshots == nil { - m.settledSnapshots = map[string]map[types.CollectionKey]bool{} + m.settledSnapshots = map[string]*snapshotRecord{} } - settled := m.settledSnapshots[gitDest.Key()] - if settled == nil { - settled = map[types.CollectionKey]bool{} - m.settledSnapshots[gitDest.Key()] = settled + record := m.settledSnapshots[gitDest.Key()] + if record == nil { + m.snapshotGenerations++ + record = &snapshotRecord{generation: m.snapshotGenerations, settled: map[types.CollectionKey]bool{}} + m.settledSnapshots[gitDest.Key()] = record } - settled[collection] = true + return record +} + +// markSnapshotSettled records that a snapshot of the collection settled: a later one of it is no +// longer initial (git.ResyncRequest.Initial). A snapshot enqueued under another generation settles +// nothing: it was taken for a record that was restarted (RestartInitialSnapshots) or forgotten since, +// and answered by a worker that may be gone. +func (m *Manager) markSnapshotSettled( + gitDest types.ResourceReference, collection types.CollectionKey, generation uint64, +) { + m.snapshotsMu.Lock() + defer m.snapshotsMu.Unlock() + record := m.settledSnapshots[gitDest.Key()] + if record == nil || record.generation != generation { + return + } + record.settled[collection] = true } // snapshotsSettled reports whether a snapshot of every one of the collections settled. func (m *Manager) snapshotsSettled(gitDest types.ResourceReference, collections []types.CollectionKey) bool { m.snapshotsMu.Lock() defer m.snapshotsMu.Unlock() - settled := m.settledSnapshots[gitDest.Key()] + var settled map[types.CollectionKey]bool + if record := m.settledSnapshots[gitDest.Key()]; record != nil { + settled = record.settled + } for _, collection := range collections { if !settled[collection] { return false @@ -80,3 +112,15 @@ func (m *Manager) forgetSettledSnapshots(gitDest types.ResourceReference) { defer m.snapshotsMu.Unlock() delete(m.settledSnapshots, gitDest.Key()) } + +// RestartInitialSnapshots starts the GitTarget's record over, under a fresh generation: every +// collection's next snapshot is initial again, and no applier on its claim is asked for anything +// until each one settled. It is for a branch worker's replacement: what settled was published by the +// old worker, into the repository it wrote, and the new worker holds none of it. A snapshot the old +// worker answers after this settles nothing. +func (m *Manager) RestartInitialSnapshots(gitDest types.ResourceReference) { + m.snapshotsMu.Lock() + defer m.snapshotsMu.Unlock() + delete(m.settledSnapshots, gitDest.Key()) + m.snapshotRecordLocked(gitDest) +} diff --git a/internal/watch/apply_state_test.go b/internal/watch/apply_state_test.go index 1f3ab212..d65d9086 100644 --- a/internal/watch/apply_state_test.go +++ b/internal/watch/apply_state_test.go @@ -33,7 +33,7 @@ func TestInitialSnapshots_TrackedPerCollection(t *testing.T) { drain := func(collection types.CollectionKey, result git.ResyncResult) { resultCh := make(chan git.ResyncResult, 1) resultCh <- result - router.drainScopedResync(gitDest, collection, "reconcile", 0, resultCh) + router.drainScopedResync(gitDest, collection, "reconcile", 0, m.snapshotGeneration(gitDest), resultCh) } drain(configmaps, git.ResyncResult{Err: git.ErrResyncDiscarded}) assert.True(t, router.initialSnapshot(gitDest, configmaps), "a discarded snapshot did not settle") @@ -52,6 +52,34 @@ func TestInitialSnapshots_TrackedPerCollection(t *testing.T) { assert.True(t, router.initialSnapshot(gitDest, configmaps), "a recreation starts over") } +// TestInitialSnapshots_AReplacedWorkerStartsOver is R4: a branch worker replaced because its +// GitProvider names another repository holds nothing the old one published, so every collection's +// next snapshot is initial again and the barrier holds until each one settled. A snapshot the old +// worker answers after the restart settles nothing: it was published into the old repository. +func TestInitialSnapshots_AReplacedWorkerStartsOver(t *testing.T) { + configmaps := types.CollectionKeyFor(schema.GroupVersionResource{Version: "v1", Resource: "configmaps"}, "") + gitDest := types.NewResourceReference("alpha", "team-a") + m := &Manager{Log: logr.Discard()} + router := &EventRouter{WatchManager: m, Log: logr.Discard()} + drain := func(generation uint64) { + resultCh := make(chan git.ResyncResult, 1) + resultCh <- git.ResyncResult{} + router.drainScopedResync(gitDest, configmaps, "reconcile", 0, generation, resultCh) + } + + drain(m.snapshotGeneration(gitDest)) + require.False(t, router.initialSnapshot(gitDest, configmaps)) + inFlight := m.snapshotGeneration(gitDest) // enqueued to the old worker, answered late + + m.RestartInitialSnapshots(gitDest) + assert.True(t, router.initialSnapshot(gitDest, configmaps), "the new worker's first snapshot is initial") + drain(inFlight) + assert.True(t, router.initialSnapshot(gitDest, configmaps), "the old worker's answer settles nothing") + + drain(m.snapshotGeneration(gitDest)) + assert.False(t, router.initialSnapshot(gitDest, configmaps), "the new worker's own snapshot settles it") +} + // TestOwnerIsReadBeforeSanitize: the applier that owns an object is stamped on its live event and on // its snapshot entry from the raw object, before sanitize strips the labels and annotations it is read // from (docs/design/applier-echoes.md, F4). diff --git a/internal/watch/event_router.go b/internal/watch/event_router.go index e5b22f06..21a22e21 100644 --- a/internal/watch/event_router.go +++ b/internal/watch/event_router.go @@ -368,17 +368,21 @@ func resyncScopeForWatchKey(key targetWatchKey) git.ResyncScope { // counts it as a background resync failure so a silently-recovered fault stays observable. The // steady-state live-event path and the next type transition recover a failed apply, so this // never re-fires the gather. +// +// snapshots is the generation of the GitTarget's settled-snapshot record the resync was enqueued +// under (Manager.snapshotGeneration): its outcome settles the collection only in that record. func (r *EventRouter) drainScopedResync( gitDest types.ResourceReference, collection types.CollectionKey, kind string, renderFidelityEpoch uint64, + snapshots uint64, resultCh chan git.ResyncResult, ) { select { case result := <-resultCh: if settlesSnapshot(result.Err) && r.WatchManager != nil { - r.WatchManager.markSnapshotSettled(gitDest, collection) + r.WatchManager.markSnapshotSettled(gitDest, collection, snapshots) } if result.Err != nil { r.handleScopedResyncError(gitDest, collection, kind, renderFidelityEpoch, result.Err) @@ -409,7 +413,7 @@ func (r *EventRouter) drainScopedResync( } case <-time.After(resyncSignalTimeout): if r.WatchManager != nil { - r.WatchManager.markSnapshotSettled(gitDest, collection) + r.WatchManager.markSnapshotSettled(gitDest, collection, snapshots) } r.Log.Error(nil, "per-type "+kind+" timed out", "gitDest", gitDest.String(), "collection", collection.String()) r.recordBackgroundResyncFailure(gitDest) diff --git a/internal/watch/event_router_test.go b/internal/watch/event_router_test.go index ff3fdbe7..513d492b 100644 --- a/internal/watch/event_router_test.go +++ b/internal/watch/event_router_test.go @@ -173,6 +173,7 @@ func TestDrainScopedResync_TreatsSupersededAsSuccess(t *testing.T) { types.CollectionKeyFor(configmapsGVR, ""), "sweep", 0, + 0, resultCh, ) }() @@ -204,6 +205,7 @@ func TestDrainScopedResync_CompletesSuccessfulResult(t *testing.T) { types.CollectionKeyFor(configmapsGVR, ""), "reconcile", 0, + 0, resultCh, ) close(done) @@ -241,7 +243,7 @@ func TestDrainScopedResync_RefusalMarksGitPathRefused(t *testing.T) { resultCh := make(chan git.ResyncResult, 1) resultCh <- git.ResyncResult{Err: fmt.Errorf("execute pending writes: %w", refusal)} - router.drainScopedResync(gitDest, key.Collection(), "reconcile", 0, resultCh) + router.drainScopedResync(gitDest, key.Collection(), "reconcile", 0, 0, resultCh) gitPath := mgr.GitPathAcceptanceForGitTarget(gitDest) @@ -567,7 +569,7 @@ func TestDrainScopedResync_QueueFullIsDrainedNotOrphaned(t *testing.T) { done := make(chan struct{}) go func() { defer close(done) - router.drainScopedResync(gitDest, collection, "reconcile", 1, resultCh) + router.drainScopedResync(gitDest, collection, "reconcile", 1, 0, resultCh) }() select { diff --git a/internal/watch/manager.go b/internal/watch/manager.go index cb1c5433..fe836ea3 100644 --- a/internal/watch/manager.go +++ b/internal/watch/manager.go @@ -111,9 +111,11 @@ type Manager struct { SourceClusters SourceClusterResolver // snapshotsMu guards settledSnapshots: per GitTarget key, the collections a snapshot of which - // the branch worker applied, refused or failed (apply_state.go). - snapshotsMu sync.Mutex - settledSnapshots map[string]map[types.CollectionKey]bool + // the branch worker applied, refused or failed, under the generation the record was started with; + // snapshotGenerations is the last generation drawn (apply_state.go). + snapshotsMu sync.Mutex + settledSnapshots map[string]*snapshotRecord + snapshotGenerations uint64 // clusters holds one clusterContext per distinct cluster — its API catalog, type registry, // and clients. configPlaneClusterID is the operator's own cluster (always present, never a diff --git a/internal/watch/materialization.go b/internal/watch/materialization.go index c61402e4..829afaef 100644 --- a/internal/watch/materialization.go +++ b/internal/watch/materialization.go @@ -83,6 +83,12 @@ func (m *Manager) ForcedRecheckTargetsForTest() []string { return forced } +// SnapshotGenerationForTest is the generation of the GitTarget's settled-snapshot record, so a test +// in another package can assert that a recovery restarted it; nothing in production calls it. +func (m *Manager) SnapshotGenerationForTest(gitDest types.ResourceReference) uint64 { + return m.snapshotGeneration(gitDest) +} + // ForgetGitTargetDeclaration drops in-memory watch state for a deleted GitTarget, and tears down // its source cluster's context when it was the last GitTarget mirroring from it. // diff --git a/internal/watch/target_watch.go b/internal/watch/target_watch.go index 5795ede6..fc5ee755 100644 --- a/internal/watch/target_watch.go +++ b/internal/watch/target_watch.go @@ -1040,6 +1040,9 @@ func (m *Manager) enqueueReplayResync( enqueued bool err error ) + // Read before the request is built: whether it is initial, and the record its outcome settles, + // are the same record (RestartInitialSnapshots). + snapshots := m.snapshotGeneration(gitDest) // A retired stream's snapshot is dropped here, at the producer: see producerGate. if !stream.gate.enqueue(ctx, func() { resultCh, enqueued, err = m.EventRouter.enqueueScopedResync( @@ -1062,7 +1065,8 @@ func (m *Manager) enqueueReplayResync( // The stream.key (GVR + namespace) is threaded to the drain for diagnostics. A refused // Git path acceptance is target-level state, so the drain records GitPathAccepted=False rather // than mutating this stream's watch readiness. - go m.EventRouter.drainScopedResync(gitDest, stream.key.Collection(), "reconcile", stream.revision, resultCh) + go m.EventRouter.drainScopedResync( + gitDest, stream.key.Collection(), "reconcile", stream.revision, snapshots, resultCh) if !enqueued { return fmt.Errorf("target replay resync for %s on %s dropped: %w", stream.key.GVR.String(), gitDest.String(), git.ErrFinalizeQueueFull) diff --git a/test/e2e/render_fidelity_e2e_test.go b/test/e2e/render_fidelity_e2e_test.go index f315e445..31ec4524 100644 --- a/test/e2e/render_fidelity_e2e_test.go +++ b/test/e2e/render_fidelity_e2e_test.go @@ -104,7 +104,34 @@ var _ = Describe("Manager Render Fidelity", Label("manager", "render-fidelity", "RenderMatchesLive", "False", "RenderDoesNotMatchLive", "${REGION}", ) - By("Flux postBuild divergence stalled the target without changing its source tree") + By("recovering once the folder is repaired in Git and a recheck is requested") + // The documented way out (docs/configuration.md): fix the folder, then annotate the + // GitTarget. The recheck re-anchors every stream, so every scope is measured afresh against + // the repaired folder, which is the only thing that clears a divergence. + run.repairSourceForm() + _, err = kubectlRunInNamespace(run.testNs, "annotate", "gittarget", run.targetName, + "reconcile.configbutler.ai/requestedAt="+time.Now().UTC().Format(time.RFC3339Nano), "--overwrite") + Expect(err).NotTo(HaveOccurred(), "failed to request a recheck of the GitTarget") + verifyResourceCondition( + "gittarget", run.targetName, run.testNs, + "RenderMatchesLive", "True", "Succeeded", "", "150s", + ) + + By("mirroring the live state the closed gate kept out, and a later edit") + Eventually(func(g Gomega) { + pullLatestRepoState(g, run.repo.CheckoutDir) + content := readRepoFile(g, filepath.Join(run.repo.CheckoutDir, renderFidelityGitPath, "deployment.yaml")) + g.Expect(content).To(ContainSubstring("value: us-east")) + g.Expect(content).To(ContainSubstring("replicas: 2")) + }, renderFidelityTimeout, resourceConditionPollInterval).Should(Succeed()) + _, err = kubectlRunInNamespace(run.testNs, "patch", "deployment", run.deploymentName, + "--type=merge", "--patch={\"spec\":{\"replicas\":3}}") + Expect(err).NotTo(HaveOccurred(), "failed to patch the live Deployment") + Eventually(func(g Gomega) { + pullLatestRepoState(g, run.repo.CheckoutDir) + content := readRepoFile(g, filepath.Join(run.repo.CheckoutDir, renderFidelityGitPath, "deployment.yaml")) + g.Expect(content).To(ContainSubstring("replicas: 3")) + }, renderFidelityTimeout, resourceConditionPollInterval).Should(Succeed()) }) }) @@ -243,6 +270,19 @@ func (r renderFidelityRun) consistentlyExpectSourceForm(seedHead string) { }, 20*time.Second, 2*time.Second).Should(Succeed()) } +// repairSourceForm resolves the postBuild token in Git, as a person repairing the folder would: the +// render then matches the live value Flux substituted. +func (r renderFidelityRun) repairSourceForm() { + Eventually(func(g Gomega) { pullLatestRepoState(g, r.repo.CheckoutDir) }).Should(Succeed()) + path := filepath.Join(r.repo.CheckoutDir, renderFidelityGitPath, "deployment.yaml") + content, err := os.ReadFile(path) + Expect(err).NotTo(HaveOccurred()) + Expect(string(content)).To(ContainSubstring("${REGION}")) + repaired := strings.ReplaceAll(string(content), "${REGION}", "us-east") + Expect(os.WriteFile(path, []byte(repaired), 0o600)).To(Succeed()) + Expect(r.checkout.commitAllAndPush("e2e: resolve the postBuild token in Git")).To(Succeed()) +} + func (r renderFidelityRun) cleanup() { cleanupWatchRule(r.watchRuleName, r.testNs) cleanupGitTarget(r.targetName, r.testNs)