feat!: recognize the applier's own writes, and flush before applying

api/v1alpha3/applytrigger_types.go

@@ -3,8 +3,6 @@
 package v1alpha3
 
 import (
-	"time"
-
 	meta "github.com/fluxcd/pkg/apis/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -63,7 +61,9 @@ const (
 	// ApplyReasonRepair puts back an API edit that an earlier apply reverted on the cluster after
 	// the edit was made. Git kept the edit; the cluster runs the older value until the repair.
 	ApplyReasonRepair ApplyReason = "Repair"
-	// ApplyReasonCatchUp keeps the applier current with GitOps Reverser's own pushes.
+	// ApplyReasonCatchUp adopts what the API created in the applier's folders, so the applier owns
+	// and can later prune it, and converges the applier once at startup and when a GitTarget first
+	// names its trigger.
 	ApplyReasonCatchUp ApplyReason = "CatchUp"
 )
 
@@ -142,9 +142,6 @@ type ApplyOperation struct {
 	Message string `json:"message,omitempty"`
 }
 
-// DefaultApplierQuietPeriod is FluxTrigger.spec.quietPeriod when it is omitted.
-const DefaultApplierQuietPeriod = 30 * time.Second
-
 // ApplyTrigger is what every trigger kind answers, so the controllers, the branch workers and the
 // dispatcher handle a FluxTrigger and an ArgoCDTrigger alike.
 // +kubebuilder:object:generate=false
@@ -160,18 +157,13 @@ type ApplyTrigger interface {
 	ApplierKind() string
 	// Applier is the object the trigger holds, its namespace defaulted to the trigger's own.
 	Applier() types.NamespacedName
-	// EffectiveQuietPeriod is how long the branch must be quiet before a CatchUp.
-	EffectiveQuietPeriod() time.Duration
 	// SourceBinding is the repository and branch the applier applies, as the trigger last read
 	// them, and false when it has not read them.
 	SourceBinding() (repositoryURL, branch string, known bool)
 	// SharedSource is the source object the applier reads, which other appliers may read too, and
 	// false when the applier has none it shares this way or the trigger has not read it. Appliers
 	// sharing one have one operation at a time.
 	SharedSource() (source types.NamespacedName, known bool)
-	// Prefetches reports whether a push may be fetched into the shared source ahead of the apply:
-	// every other reader of it is held, so a new artifact applies nothing on its own.
-	Prefetches() bool
 	// Operation is status.operation, and SetOperation replaces it and status.repairPending.
 	Operation() *ApplyOperation
 	RepairPending() bool

api/v1alpha3/argocdtrigger_types.go

@@ -4,16 +4,12 @@ package v1alpha3
 
 import (
 	"strings"
-	"time"
 
 	meta "github.com/fluxcd/pkg/apis/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 )
 
-// Why no quietPeriod: it times a CatchUp, and an Application is never asked to catch up. With
-// automated sync off it applies nothing on its own, so it holds no stale state to correct.
-
 // ArgoCDTriggerSpec names the Argo CD Application the trigger holds, and the cluster it is in.
 type ArgoCDTriggerSpec struct {
 	// Immutable for the reason FluxTriggerSpec's references are: the trigger holds what it names.
@@ -77,7 +73,7 @@ type ArgoCDTriggerStatus struct {
 	Operation *ApplyOperation `json:"operation,omitempty"`
 
 	// RepairPending is true while a Repair sync is pending for the Application: a sync reverted an
-	// API edit made after it was asked for, so the cluster runs an older value than Git until the
+	// API edit made after its flush, so the cluster runs an older value than Git until the
 	// next sync.
 	// +optional
 	RepairPending bool `json:"repairPending,omitempty"`
@@ -108,9 +104,9 @@ type ArgoCDTriggerSource struct {
 // +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
 
 // ArgoCDTrigger holds an Argo CD Application, with automated sync off, for GitOps Reverser, which
-// starts every sync it runs: only on a branch with nothing live missing from the commit, after a
-// commit GitOps Reverser did not push, after a refused edit, and after a reverted edit. GitTargets
-// name it with spec.applyTriggerRef.
+// starts every sync it runs, each after publishing the edits pending on the branch: after a commit
+// GitOps Reverser did not push, after a refused edit, after a reverted edit, and to adopt what the
+// API created or converge the Application (CatchUp). GitTargets name it with spec.applyTriggerRef.
 type ArgoCDTrigger struct {
 	metav1.TypeMeta `json:",inline"`
 
@@ -163,9 +159,6 @@ func (t *ArgoCDTrigger) Applier() types.NamespacedName {
 	}
 }
 
-// EffectiveQuietPeriod is the default: it only times a CatchUp, which an Application is never asked.
-func (t *ArgoCDTrigger) EffectiveQuietPeriod() time.Duration { return DefaultApplierQuietPeriod }
-
 // SourceBinding is status.source's repoURL, and the branch its targetRevision names ("HEAD" for an
 // empty one, which follows no branch GitOps Reverser writes).
 func (t *ArgoCDTrigger) SourceBinding() (string, string, bool) {
@@ -185,9 +178,6 @@ func (t *ArgoCDTrigger) SharedSource() (types.NamespacedName, bool) {
 	return types.NamespacedName{}, false
 }
 
-// Prefetches is false: an Application has no source object to fetch ahead.
-func (t *ArgoCDTrigger) Prefetches() bool { return false }
-
 // Operation is status.operation.
 func (t *ArgoCDTrigger) Operation() *ApplyOperation { return t.Status.Operation }
 

api/v1alpha3/fluxtrigger_types.go

@@ -3,8 +3,6 @@
 package v1alpha3
 
 import (
-	"time"
-
 	meta "github.com/fluxcd/pkg/apis/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -43,16 +41,6 @@ type FluxTriggerSpec struct {
 	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="spec.kustomizationRef is immutable; delete and recreate the FluxTrigger to hold another Kustomization"
 	// +kubebuilder:validation:XValidation:rule="self.name != ''",message="spec.kustomizationRef.name must not be empty"
 	KustomizationRef meta.NamespacedObjectReference `json:"kustomizationRef"`
-
-	// QuietPeriod is how long the branch must hold no open commit window and no pending writes,
-	// after GitOps Reverser's own pushes, before the Kustomization is asked to catch up with them.
-	// Every new write restarts the wait. A Go duration of at most "1h".
-	// +optional
-	// +kubebuilder:default="30s"
-	// +kubebuilder:validation:Type=string
-	// +kubebuilder:validation:Pattern="^([0-9]+(\\.[0-9]+)?(ns|us|µs|μs|ms|s|m|h))+$"
-	// +kubebuilder:validation:XValidation:rule="duration(self) <= duration('1h')",message="spec.quietPeriod must be a Go duration of at most 1h"
-	QuietPeriod *metav1.Duration `json:"quietPeriod,omitempty"`
 }
 
 // FluxTriggerStatus is what the trigger knows about its Kustomization, and the apply it started.
@@ -93,7 +81,7 @@ type FluxTriggerStatus struct {
 	Operation *ApplyOperation `json:"operation,omitempty"`
 
 	// RepairPending is true while a Repair apply is pending for the Kustomization: an apply reverted
-	// an API edit made after it was asked for, so the cluster runs an older value than Git until the
+	// an API edit made after its flush, so the cluster runs an older value than Git until the
 	// next apply.
 	// +optional
 	RepairPending bool `json:"repairPending,omitempty"`
@@ -123,13 +111,6 @@ type FluxTriggerSource struct {
 	// +optional
 	// +listType=atomic
 	OtherConsumers []meta.NamespacedObjectKindReference `json:"otherConsumers,omitempty"`
-
-	// Prefetch is true while GitOps Reverser fetches the GitRepository as soon as it learns of a push,
-	// before the apply: every consumer of it was listed and is held by a FluxTrigger, so a new
-	// artifact applies nothing on its own. False when otherConsumers is not empty, or the consumers
-	// could not be listed (the identity needs list on kustomizations, helmreleases and helmcharts).
-	// +optional
-	Prefetch bool `json:"prefetch,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -145,10 +126,10 @@ type FluxTriggerSource struct {
 // +kubebuilder:printcolumn:name="Message",type=string,JSONPath=`.status.conditions[?(@.type=="Ready")].message`,priority=1
 // +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
 
-// FluxTrigger holds a Flux Kustomization for GitOps Reverser, which starts every apply it makes:
-// only on a branch with nothing live missing from the commit, after a commit GitOps Reverser did
-// not push, after a refused edit, after a reverted edit, and to catch up with its own pushes.
-// GitTargets name it with spec.applyTriggerRef.
+// FluxTrigger holds a Flux Kustomization for GitOps Reverser, which starts every apply it makes,
+// each after publishing the edits pending on the branch: after a commit GitOps Reverser did not
+// push, after a refused edit, after a reverted edit, and to adopt what the API created or converge
+// the Kustomization (CatchUp). GitTargets name it with spec.applyTriggerRef.
 type FluxTrigger struct {
 	metav1.TypeMeta `json:",inline"`
 
@@ -199,14 +180,6 @@ func (t *FluxTrigger) Applier() types.NamespacedName {
 	}
 }
 
-// EffectiveQuietPeriod is spec.quietPeriod, resolving an omitted field to the default.
-func (t *FluxTrigger) EffectiveQuietPeriod() time.Duration {
-	if t.Spec.QuietPeriod == nil {
-		return DefaultApplierQuietPeriod
-	}
-	return t.Spec.QuietPeriod.Duration
-}
-
 // SourceBinding is status.source's url and branch.
 func (t *FluxTrigger) SourceBinding() (string, string, bool) {
 	if t.Status.Source == nil {
@@ -224,9 +197,6 @@ func (t *FluxTrigger) SharedSource() (types.NamespacedName, bool) {
 	return types.NamespacedName{Namespace: source.Namespace, Name: source.Name}, true
 }
 
-// Prefetches is status.source.prefetch.
-func (t *FluxTrigger) Prefetches() bool { return t.Status.Source != nil && t.Status.Source.Prefetch }
-
 // Operation is status.operation.
 func (t *FluxTrigger) Operation() *ApplyOperation { return t.Status.Operation }
 

api/v1alpha3/gittarget_types.go

@@ -92,11 +92,12 @@ type GitTargetSpec struct {
 
 	// ApplyTriggerRef names the apply trigger, in this GitTarget's namespace, that holds the applier
 	// of this folder: a FluxTrigger (a Flux Kustomization) or an ArgoCDTrigger (an Argo CD
-	// Application). With it set, GitOps Reverser decides when that applier applies: only when the
-	// branch holds no open commit window and no pending writes, after a commit it did not push,
-	// after a refused edit (spec.onRefusal: Reapply), after an apply reverted a newer API edit, and,
-	// for a Kustomization, once the branch has been quiet for the trigger's quietPeriod, to keep it
-	// current with its own pushes. The trigger's applier must apply this target's repository and
+	// Application). With it set, GitOps Reverser decides when that applier applies, always after
+	// publishing the edits it holds: after a commit it did not push, after a refused edit
+	// (spec.onRefusal: Reapply), after an apply reverted a newer API edit, after the API created an
+	// object in the folder (so the applier owns it and prunes it later), and once when the operator
+	// starts or the trigger is first named, after the folder's initial snapshots are in Git. Objects
+	// another applier owns are not mirrored into the folder. The trigger's applier must apply this target's repository and
 	// branch; the ApplyTriggerReady condition says whether it does. Several GitTargets may name one
 	// trigger, and it is asked once for all of them.
 	// +optional
@@ -109,8 +110,8 @@ type GitTargetSpec struct {
 	// "Ignore", the default, records the refusal in conditions and events and does nothing else.
 	//
 	// "Reapply" asks this target's applier (through spec.applyTriggerRef, which it requires) to
-	// apply the current tip once the branch holds no open window and no pending writes, which
-	// reverts the refused edit and nothing that is still waiting to be published. Only this
+	// apply the current tip after publishing the edits pending on the branch, which reverts the
+	// refused edit and nothing that was waiting to be published. Only this
 	// target's applier is asked. If its GitRepository has to fetch to reach the tip, the
 	// Kustomizations a FluxTrigger holds stay suspended and ignore the new artifact; only the
 	// consumers no trigger holds, listed in the trigger's status.source.otherConsumers, apply it, as
@@ -447,9 +448,9 @@ type GitTargetStatus struct {
 
 // GitTargetRemoteStatus is the answer to "where is my branch, and when did we last prove it".
 //
-// It is written whenever the commit changes — including one we pushed ourselves — and otherwise
-// only when the published timestamp is older than one refresh interval: the commit has to move
-// with the fact it dates.
+// It is sampled when the GitTarget reconciles, from what its branch worker last proved about the
+// branch, and written at most once a minute. The first observation, and a correction after the
+// repository changed under the target, are written at once.
 type GitTargetRemoteStatus struct {
 	// Commit is the commit hash the branch is at on the remote. EMPTY means the branch is not on
 	// the remote at all, which is not an error: a branch does not exist without a commit, and a

api/v1alpha3/helpers_test.go

@@ -3,8 +3,6 @@
 package v1alpha3
 
 import (
-	"time"
-
 	"testing"
 
 	meta "github.com/fluxcd/pkg/apis/meta"
@@ -336,7 +334,6 @@ func TestApplyTrigger_AppliersDefaultToTheTriggersNamespace(t *testing.T) {
 	}
 	assert.Equal(t, "team-a/apps", flux.Applier().String())
 	assert.Equal(t, "hub", flux.ClusterProvider())
-	assert.Equal(t, DefaultApplierQuietPeriod, flux.EffectiveQuietPeriod())
 
 	argo := &ArgoCDTrigger{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "team-a"},
@@ -364,27 +361,23 @@ func TestArgoCDTrigger_SourceBindingNamesTheBranch(t *testing.T) {
 }
 
 // TestNewApplyTrigger_KnowsBothKinds: a kind this release does not know is nil, never a guess.
-// TestApplyTrigger_SharedSource: a FluxTrigger shares the GitRepository it last read, and prefetches
-// only when its status says so; an ArgoCDTrigger shares nothing and never prefetches.
+// TestApplyTrigger_SharedSource: a FluxTrigger shares the GitRepository it last read; an
+// ArgoCDTrigger shares nothing.
 func TestApplyTrigger_SharedSource(t *testing.T) {
 	flux := &FluxTrigger{}
 	_, known := flux.SharedSource()
 	assert.False(t, known, "a trigger that has not read its source shares none yet")
-	assert.False(t, flux.Prefetches())
 
 	flux.Status.Source = &FluxTriggerSource{
 		GitRepository: meta.NamespacedObjectReference{Namespace: "flux-system", Name: "platform"},
-		Prefetch:      true,
 	}
 	source, known := flux.SharedSource()
 	assert.True(t, known)
 	assert.Equal(t, "flux-system/platform", source.String())
-	assert.True(t, flux.Prefetches())
 
 	argo := &ArgoCDTrigger{}
 	_, known = argo.SharedSource()
 	assert.False(t, known, "every sync pins its own revision")
-	assert.False(t, argo.Prefetches())
 }
 
 func TestNewApplyTrigger_KnowsBothKinds(t *testing.T) {
@@ -397,8 +390,7 @@ func TestNewApplyTrigger_KnowsBothKinds(t *testing.T) {
 // and SetOperation replaces the operation and repairPending together.
 func TestApplyTrigger_StatusAccessors(t *testing.T) {
 	operation := &ApplyOperation{Phase: ApplyOperationRunning, Commit: "abc"}
-	quiet := &metav1.Duration{Duration: time.Minute}
-	flux := &FluxTrigger{Spec: FluxTriggerSpec{QuietPeriod: quiet}}
+	flux := &FluxTrigger{}
 	argo := &ArgoCDTrigger{}
 	for _, trigger := range []ApplyTrigger{flux, argo} {
 		assert.Nil(t, trigger.Operation())
@@ -414,10 +406,8 @@ func TestApplyTrigger_StatusAccessors(t *testing.T) {
 
 	assert.Equal(t, ApplyTriggerKindFlux, flux.TriggerKind())
 	assert.Equal(t, ApplierKindKustomization, flux.ApplierKind())
-	assert.Equal(t, time.Minute, flux.EffectiveQuietPeriod())
 	assert.Equal(t, ApplyTriggerKindArgoCD, argo.TriggerKind())
 	assert.Equal(t, ApplierKindApplication, argo.ApplierKind())
-	assert.Equal(t, DefaultApplierQuietPeriod, argo.EffectiveQuietPeriod())
 	assert.Equal(t, DefaultClusterProviderName, argo.ClusterProvider(),
 		"an omitted clusterProviderRef is the default provider, as for a GitTarget")
 	argo.Spec.ClusterProviderRef = &meta.LocalObjectReference{Name: "hub"}

charts/gitops-reverser/templates/applier-rbac.yaml

@@ -2,7 +2,7 @@
 # What GitOps Reverser needs to drive a Kustomization a FluxTrigger holds: read and patch the
 # Kustomization and its GitRepository (it suspends and resumes the one and asks the other to fetch),
 # list the GitRepository's consumers (Kustomizations, and HelmReleases with the HelmCharts built
-# from it; without it the source is never prefetched), and create SelfSubjectAccessReviews so the operator can check it holds that
+# from it; without it the trigger lists none in status.source.otherConsumers), and create SelfSubjectAccessReviews so the operator can check it holds that
 # access before relying on it. Separate from the manager role because it is optional and
 # names another project's API: an install without Flux grants nothing on it. See
 # docs/design/apply-trigger.md.

charts/gitops-reverser/values.yaml

@@ -302,8 +302,9 @@ attribution:
   # bounds stream retention and the in-memory index together, and doubles as the replay horizon a
   # restart warms the index from.
   ttl: "10m"
-  # Bounded per-event wait for a matching audit fact before a watch event ships as the committer.
-  # Larger values raise attribution hit-rate at the cost of commit latency.
+  # Bounded per-event wait for a matching audit fact before a watch event ships unresolved, authored
+  # "unknown (attribution unresolved)". Larger values raise attribution hit-rate at the cost of
+  # commit latency.
   grace: "3s"
   # Caps on the facts held in memory. Per-type is the primary because it is the fair one: a burst on
   # one noisy type — a deletecollection over ten thousand objects, a large rollout — must not evict

cmd/main.go

@@ -181,6 +181,9 @@ func main() {
 	// The read-only twin of the refusal reporter: the refresher re-reads an idle target's folder,
 	// and a folder somebody else broke in Git produces no Kubernetes event to notice it by.
 	workerManager.SetScanAcceptanceReporter(watchMgr.ReportGitPathScan)
+	// A worker discards a snapshot an apply made stale and asks for a fresh one, and asks no applier
+	// for anything before a GitTarget's initial snapshots are in Git: both read the streams.
+	workerManager.SetSnapshotHooks(watchMgr.InitialSnapshotsSettled, watchMgr.RequestRecheckForGitTarget)
 	// The branch workers decide when a GitTarget's applier may apply; the dispatcher delivers it,
 	// off their queues, to the applier its apply trigger holds, in the trigger's cluster. The status
 	// writer records each operation on the trigger, also off their queues.
@@ -614,9 +617,9 @@ func parseFlagsWithArgs(fs *flag.FlagSet, args []string) (appConfig, error) {
 			"decides how often the precise path is taken; drops are counted on "+
 			"attribution_collection_without_uidset_total{reason}.")
 	fs.DurationVar(&cfg.attributionGrace, "author-attribution-grace", watch.DefaultAttributionGraceWindow,
-		"Bounded per-event wait for a matching audit fact to arrive before a watch event ships as the "+
-			"configured committer (duration string; default 3s). Larger values raise attribution hit-rate "+
-			"at the cost of commit latency.")
+		"Bounded per-event wait for a matching audit fact to arrive before a watch event ships unresolved, "+
+			"authored \"unknown (attribution unresolved)\" (duration string; default 3s). Larger values "+
+			"raise attribution hit-rate at the cost of commit latency.")
 	fs.DurationVar(&cfg.commitRequestTTL, "commit-request-ttl", controller.DefaultCommitRequestTTL,
 		"How long a CommitRequest is kept once it finishes (duration string; default 48h, 0 keeps "+
 			"them). The deletion time is written onto the request as the "+