fix(providers): bind pre-upgrade ledger receipts from live evidence

app/operator/config/crd/bases/dns.appthrust.io_zoneunits.yaml

@@ -499,8 +499,10 @@ spec:
                     recordSetUID:
                       description: |-
                         RecordSetUID identifies the claim incarnation observed by the provider.
-                        Empty or mismatched UIDs cannot authorize cleanup or transfer provider
-                        ownership to a recreated claim, even when name and generation match.
+                        A mismatched UID cannot authorize cleanup or transfer provider ownership
+                        to a recreated claim, even when name and generation match. An empty UID
+                        is a pre-upgrade receipt: it binds to the current claim only when the
+                        provider observes the live record equal to the claim's desired state.
                       minLength: 1
                       type: string
                   required:

deploy/charts/dns-api/crds/dns.appthrust.io_zoneunits.yaml

@@ -499,8 +499,10 @@ spec:
                     recordSetUID:
                       description: |-
                         RecordSetUID identifies the claim incarnation observed by the provider.
-                        Empty or mismatched UIDs cannot authorize cleanup or transfer provider
-                        ownership to a recreated claim, even when name and generation match.
+                        A mismatched UID cannot authorize cleanup or transfer provider ownership
+                        to a recreated claim, even when name and generation match. An empty UID
+                        is a pre-upgrade receipt: it binds to the current claim only when the
+                        provider observes the live record equal to the claim's desired state.
                       minLength: 1
                       type: string
                   required:

docs/design/core-api.md

@@ -729,7 +729,7 @@ If multiple `RecordSet` claims target the same record identity, core keeps the e
 - `recordSets`: provider result for accepted owner `RecordSet` claims.
 - `recordSets[].recordSetNamespace`: owner `RecordSet` namespace matching `spec.recordSets[]`.
 - `recordSets[].recordSetName`: owner `RecordSet` name matching `spec.recordSets[]`.
-- `recordSets[].recordSetUID`: exact nonempty UID of the observed spec item. Missing or different UIDs do not establish ownership or authorize claim finalization.
+- `recordSets[].recordSetUID`: exact nonempty UID of the observed spec item. A different UID never establishes ownership or authorizes claim finalization; a missing UID is a pre-upgrade receipt that binds only through the live evidence described under **UID cutover and legacy state**.
 - `recordSets[].observedGeneration`: owner `RecordSet` generation observed in `ZoneUnit.spec`.
 - `recordSets[].conditions`: provider acceptance and programming result for that `RecordSet` claim.
 - `recordSets[].provider.data`: public provider data to project to `RecordSet.status.provider.data`.
@@ -746,11 +746,15 @@ Route 53 pending affected-record entries include the claim UID. A legacy or pred
 
 **UID cutover and legacy state.**
 
-The UID fields are optional on the wire so existing ledgers remain readable. Explicit empty strings are invalid in the CRD; omitted or mismatched UIDs still fail closed in controllers. There is no namespace/name/generation fallback and no automatic relabeling of old provider state.
+The UID fields are optional on the wire so existing ledgers remain readable. Explicit empty strings are invalid in the CRD; a mismatched UID always fails closed in controllers. There is no namespace/name/generation fallback and no relabeling of old provider state by name alone.
 
-This is not a transparent in-place upgrade for existing UID-less ownership receipts. Apply the updated CRDs before adopting the corresponding controllers; Helm does not update existing CRDs from a chart's `crds/` directory during a normal upgrade. Qualify the coordinated controller/schema cutover and recovery of existing claims separately.
+A `status.recordSets[]` entry without `recordSetUID` is a pre-upgrade receipt: the provider's own record that it programmed the record for that claim name, without proof of which incarnation. Providers bind such a receipt to the current claim incarnation only from live evidence, in place and without a provider write: the receipt must name exactly the record identity the claim desires (Route 53 hosted zone, record name and type; Cloudflare record IDs covering every same-type record at that name) and the live record must already equal the claim's desired state. The binding then only records which incarnation owns the record; a live record that differs from desired stays `ProviderConflict`, because a pre-upgrade receipt cannot distinguish an ordinary desired-value update from a recreated claim inheriting a stranger's record. The same evidence authorizes deletion for a deleting claim whose retained desired values still match the live record; a minimal cleanup item without values cannot match. A receipt marked `deletionCompleted` never binds. Providers emit a `*LegacyOwnershipBound` event when a receipt is bound.
 
-Already-present external records with unbound ownership remain conflicts. Recovery uses separately authorized, provider-validated explicit `RecordSet.spec.adoption`, or a fresh provider observation establishing absence before normal creation or cleanup. Desired-value equality alone does not adopt a foreign Cloudflare record ID. Do not patch UIDs, provider status, or finalizers to manufacture ownership, and do not enable zone-wide adoption as an automatic fallback. The source correction neither restores already-lost live authority nor authorizes release or deployment.
+Apply the updated CRDs before adopting the corresponding controllers; Helm does not update existing CRDs from a chart's `crds/` directory during a normal upgrade, so the CRD delivery owner must apply them with the controller image (Argo CD renders `crds/` and upgrades them unless `skipCrds` is set). Until the CRD carries `recordSetUID`, the field is pruned from `ZoneUnit.spec` and every provider status write fails with a source-changed conflict.
+
+Providers also prune orphaned ledger entries whose claim namespace/name has no `spec.recordSets[]` item once the entry can no longer matter: `deletionCompleted` is set, or (Route 53) the record it names is absent from the hosted zone, or (Cloudflare) it records no IDs. An orphan that still names a live record is kept deliberately: core may drop a spec item transiently (a rejected claim whose generation changed) and re-add it later, and the retained entry is the only evidence of a leaked record. Cloudflare does not confirm absence per ID, so ID-bearing orphans stay until an operator removes the records.
+
+Already-present external records with unbound ownership and no matching receipt remain conflicts. Recovery uses separately authorized, provider-validated explicit `RecordSet.spec.adoption`, or a fresh provider observation establishing absence before normal creation or cleanup. Desired-value equality alone does not adopt a foreign Cloudflare record ID. Do not patch UIDs, provider status, or finalizers to manufacture ownership, and do not enable zone-wide adoption as an automatic fallback. The source correction neither restores already-lost live authority nor authorizes release or deployment.
 
 Provider-specific status follows a public-data boundary:
 

internal/go/providers/cloudflare/controllers/zoneunit/recordset.go

@@ -73,10 +73,14 @@ type recordSetReconciler struct {
 }
 
 func (r *recordSetReconciler) reconcileZoneUnitRecordSets(ctx context.Context, unit *dnsv1alpha1.ZoneUnit) (ctrl.Result, error) {
+	if pruned, err := r.pruneOrphanedRecordSetStatuses(ctx, unit); err != nil || pruned {
+		return ctrl.Result{Requeue: pruned}, err
+	}
 	statusByRef := map[string]dnsv1alpha1.ZoneUnitRecordSetStatus{}
 	for _, status := range unit.Status.RecordSets {
 		statusByRef[cloudflareRecordSetClaimKey(status.RecordSetNamespace, status.RecordSetName)] = status
 	}
+	legacy := legacyRecordSetReceipts(unit)
 
 	var aggregate ctrl.Result
 	for _, item := range unit.Spec.RecordSets {
@@ -87,9 +91,13 @@ func (r *recordSetReconciler) reconcileZoneUnitRecordSets(ctx context.Context, u
 			aggregate = mergeCloudflareRecordSetResult(aggregate, ctrl.Result{RequeueAfter: r.requeueAfter()})
 			continue
 		}
-		recordSet := cloudflareRecordSetFromZoneUnitItem(unit, item, statusByRef[cloudflareRecordSetClaimKey(item.RecordSetNamespace, item.RecordSetName)])
+		claimKey := cloudflareRecordSetClaimKey(item.RecordSetNamespace, item.RecordSetName)
+		recordSet := cloudflareRecordSetFromZoneUnitItem(unit, item, statusByRef[claimKey])
 		ctxData, accepted, err := r.acceptRecordSetFromZoneUnit(ctx, &recordSet, unit)
 		ctxData.RecordSetSource = item.DeepCopy()
+		if receipt, ok := legacy[claimKey]; ok {
+			ctxData.LegacyReceipt = &receipt
+		}
 		if err != nil || !accepted {
 			return aggregate, err
 		}
@@ -130,8 +138,11 @@ type cloudflareRecordSetContext struct {
 	Version         *dnsv1alpha1.ProviderVersion
 	ZoneUnit        *dnsv1alpha1.ZoneUnit
 	RecordSetSource *dnsv1alpha1.ZoneUnitRecordSetSpec
-	ZoneID          string
-	FullName        string
+	// LegacyReceipt is the UID-less pre-upgrade ledger entry for this claim,
+	// if any. It is never loaded into RecordSet status; see legacyReceiptRecords.
+	LegacyReceipt *dnsv1alpha1.ZoneUnitRecordSetStatus
+	ZoneID        string
+	FullName      string
 }
 
 func cloudflareRecordSetFromZoneUnitItem(unit *dnsv1alpha1.ZoneUnit, item dnsv1alpha1.ZoneUnitRecordSetSpec, status dnsv1alpha1.ZoneUnitRecordSetStatus) dnsv1alpha1.RecordSet {
@@ -369,6 +380,12 @@ func (r *recordSetReconciler) reconcileNormal(ctx context.Context, provider Reco
 	managedIDs := cloudflareStatusRecordIDs(recordSet)
 	if len(managedIDs) == 0 {
 		if len(sameType) > 0 {
+			if ctxData.LegacyReceipt != nil {
+				if bound, ok := legacyReceiptRecords(*ctxData.LegacyReceipt, recordSet, ctxData.FullName, current); ok {
+					r.recordEvent(recordSet, corev1.EventTypeNormal, "CloudflareRecordSetLegacyOwnershipBound", fmt.Sprintf("pre-upgrade Cloudflare ownership receipt was bound to RecordSet %s", recordSet.UID))
+					return ctrl.Result{}, r.setReady(ctx, recordSet, bound)
+				}
+			}
 			return ctrl.Result{}, r.setProgrammed(ctx, recordSet, metav1.ConditionFalse, "ProviderConflict", "same Cloudflare DNS records already exist")
 		}
 		return r.applyCloudflareRecordSetDiff(ctx, provider, recordSet, ctxData, nil, desired)
@@ -444,10 +461,22 @@ func (r *recordSetReconciler) reconcileDeleteWithContext(ctx context.Context, re
 		if err != nil {
 			return r.failProgrammedForProviderError(ctx, recordSet, err)
 		}
-		if len(observed) > 0 {
+		if len(observed) == 0 {
+			return ctrl.Result{}, r.removeFinalizer(ctx, recordSet)
+		}
+		bound, err := r.bindLegacyReceiptForDeletion(ctx, recordSet, ctxData, observed)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		if !bound {
 			return ctrl.Result{}, r.setProgrammed(ctx, recordSet, metav1.ConditionFalse, "ProviderOwnershipNotEstablished", "Cloudflare DNS records exist but are not owned by the current RecordSet UID")
 		}
-		return ctrl.Result{}, r.removeFinalizer(ctx, recordSet)
+		// patchStatus rebased recordSet.Status onto the bound ledger entry.
+		statusData, err = cloudflareRecordSetStatusData(recordSet)
+		if err != nil {
+			return ctrl.Result{}, r.setProgrammed(ctx, recordSet, metav1.ConditionFalse, "ReconcileError", err.Error())
+		}
+		ids = cloudflareRecordStatusIDs(statusData.Records)
 	}
 	var deletes []CloudflareDNSRecord
 	for _, id := range ids {

internal/go/providers/cloudflare/controllers/zoneunit/recordset_ledger.go

@@ -0,0 +1,142 @@
+package cloudflare
+
+import (
+	"context"
+	"fmt"
+	"slices"
+
+	cloudflarev1alpha1 "github.com/appthrust/dns-api/pkg/go/api/cloudflare/v1alpha1"
+	dnsv1alpha1 "github.com/appthrust/dns-api/pkg/go/api/dns/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// legacyRecordSetReceipts returns the ZoneUnit.status.recordSets[] entries
+// written before recordSetUID existed, keyed by claim namespace/name.
+//
+// Such an entry is the provider's own receipt that it created the listed
+// record IDs for that claim name, but it cannot prove which claim incarnation
+// they belong to. Callers must pair it with fresh live evidence before treating
+// the records as owned; a completed or ID-less entry carries no receipt.
+func legacyRecordSetReceipts(unit *dnsv1alpha1.ZoneUnit) map[string]dnsv1alpha1.ZoneUnitRecordSetStatus {
+	receipts := map[string]dnsv1alpha1.ZoneUnitRecordSetStatus{}
+	for _, status := range unit.Status.RecordSets {
+		if status.RecordSetUID != "" || status.DeletionCompleted {
+			continue
+		}
+		data, err := cloudflareRecordSetStatusDataFromProvider(status.Provider)
+		if err != nil || len(cloudflareRecordStatusIDs(data.Records)) == 0 {
+			continue
+		}
+		receipts[cloudflareRecordSetClaimKey(status.RecordSetNamespace, status.RecordSetName)] = status
+	}
+	return receipts
+}
+
+// legacyReceiptRecords returns the live records a UID-less receipt may bind to
+// the current claim incarnation. Every listed record of the claim's type must
+// carry an ID from the receipt, and those records must already equal the
+// desired state: the migration then changes nothing in Cloudflare, it only
+// records which incarnation owns the IDs. A foreign same-type record or a live
+// value that differs from desired stays a conflict, because a pre-upgrade
+// receipt cannot distinguish an ordinary desired-value update from a recreated
+// claim inheriting a stranger's records.
+func legacyReceiptRecords(receipt dnsv1alpha1.ZoneUnitRecordSetStatus, recordSet *dnsv1alpha1.RecordSet, fullName string, listed []CloudflareDNSRecord) ([]CloudflareDNSRecord, bool) {
+	data, err := cloudflareRecordSetStatusDataFromProvider(receipt.Provider)
+	if err != nil {
+		return nil, false
+	}
+	ids := cloudflareRecordStatusIDs(data.Records)
+	if len(ids) == 0 {
+		return nil, false
+	}
+	options, err := cloudflareRecordSetOptions(recordSet)
+	if err != nil || validateCloudflareRecordSetOptions(recordSet, options) != "" {
+		return nil, false
+	}
+	desired, err := desiredCloudflareDNSRecords(recordSet, fullName, options)
+	if err != nil {
+		return nil, false
+	}
+	sameType := cloudflareRecordsByType(listed, string(recordSet.Spec.Type))
+	if len(sameType) == 0 {
+		return nil, false
+	}
+	for _, record := range sameType {
+		if !slices.Contains(ids, record.ID) {
+			return nil, false
+		}
+	}
+	if !cloudflareDNSRecordSetEqual(sameType, desired) {
+		return nil, false
+	}
+	return sameType, true
+}
+
+// bindLegacyReceiptForDeletion binds a pre-upgrade receipt to a deleting claim
+// when the retained desired values still match the live records. Deletion is
+// the one mutation a claim's own receipt must keep authorizing after the UID
+// cutover; otherwise every pre-upgrade claim would be stuck in deletion. A
+// minimal cleanup item without values cannot match and stays a conflict.
+func (r *recordSetReconciler) bindLegacyReceiptForDeletion(ctx context.Context, recordSet *dnsv1alpha1.RecordSet, ctxData cloudflareRecordSetContext, observed []CloudflareDNSRecord) (bool, error) {
+	if ctxData.LegacyReceipt == nil {
+		return false, nil
+	}
+	bound, ok := legacyReceiptRecords(*ctxData.LegacyReceipt, recordSet, ctxData.FullName, observed)
+	if !ok {
+		return false, nil
+	}
+	if err := r.patchCloudflareRecordSetStatus(ctx, recordSet, func(data *cloudflarev1alpha1.CloudflareRecordSetStatusData, state *cloudflarev1alpha1.CloudflareRecordSetState) {
+		data.Records = cloudflareDNSRecordStatuses(bound)
+		state.Records = cloudflareDNSRecordStates(bound)
+	}); err != nil {
+		return false, err
+	}
+	r.recordEvent(recordSet, corev1.EventTypeNormal, "CloudflareRecordSetLegacyOwnershipBound", fmt.Sprintf("pre-upgrade Cloudflare ownership receipt was bound to deleting RecordSet %s", recordSet.UID))
+	return true, nil
+}
+
+// pruneOrphanedRecordSetStatuses drops status.recordSets[] entries whose claim
+// key has no spec item and which can no longer matter: cleanup completed, or
+// no record ID is recorded. Without this the ledger grows with every deletion
+// and pre-upgrade UID-less entries stay forever.
+//
+// An orphan that still records IDs is kept on purpose. Core can drop a spec
+// item transiently (a claim rejected while its generation changed) and re-add
+// it later; pruning that receipt would turn the provider's own records into a
+// permanent ProviderConflict. Confirming absence would need one Cloudflare
+// read per recorded ID on every reconcile, so the retained entry stays as the
+// only evidence of a leaked record.
+//
+// Returns true after a write so the caller re-observes the ZoneUnit. The
+// caller's unit may already trail zone-status writes from the same pass, so
+// the patch is based on a fresh observation rather than that snapshot.
+func (r *recordSetReconciler) pruneOrphanedRecordSetStatuses(ctx context.Context, unit *dnsv1alpha1.ZoneUnit) (bool, error) {
+	var observed dnsv1alpha1.ZoneUnit
+	if err := r.Get(ctx, client.ObjectKeyFromObject(unit), &observed); err != nil {
+		return false, client.IgnoreNotFound(err)
+	}
+	claims := make(map[string]struct{}, len(observed.Spec.RecordSets))
+	for _, item := range observed.Spec.RecordSets {
+		claims[cloudflareRecordSetClaimKey(item.RecordSetNamespace, item.RecordSetName)] = struct{}{}
+	}
+	prunable := func(status dnsv1alpha1.ZoneUnitRecordSetStatus) bool {
+		if _, claimed := claims[cloudflareRecordSetClaimKey(status.RecordSetNamespace, status.RecordSetName)]; claimed {
+			return false
+		}
+		if status.DeletionCompleted {
+			return true
+		}
+		data, err := cloudflareRecordSetStatusDataFromProvider(status.Provider)
+		return err != nil || len(cloudflareRecordStatusIDs(data.Records)) == 0
+	}
+	if !slices.ContainsFunc(observed.Status.RecordSets, prunable) {
+		return false, nil
+	}
+	base := observed.DeepCopy()
+	observed.Status.RecordSets = slices.DeleteFunc(observed.Status.RecordSets, prunable)
+	if err := r.Status().Patch(ctx, &observed, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil {
+		return false, client.IgnoreNotFound(err)
+	}
+	return true, nil
+}