From d62a4a1569500dcffc018317e826397bcc9ae237 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Mon, 12 Oct 2026 12:09:16 -0500 Subject: [PATCH 1/3] feat: Distribute private DNS configuration across regional fleets --- internal/internaldns/platform/acks.go | 551 ++++++++++++++++++ internal/internaldns/platform/outbox.go | 339 +++++++++++ internal/internaldns/platform/planner.go | 273 +++++++++ .../internaldns/platform/platform_test.go | 457 +++++++++++++++ internal/internaldns/platform/values.go | 12 + 5 files changed, 1632 insertions(+) create mode 100644 internal/internaldns/platform/acks.go create mode 100644 internal/internaldns/platform/outbox.go create mode 100644 internal/internaldns/platform/planner.go create mode 100644 internal/internaldns/platform/platform_test.go create mode 100644 internal/internaldns/platform/values.go diff --git a/internal/internaldns/platform/acks.go b/internal/internaldns/platform/acks.go new file mode 100644 index 00000000..28b6fd1c --- /dev/null +++ b/internal/internaldns/platform/acks.go @@ -0,0 +1,551 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package platform + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" + "go.miloapis.com/dns-operator/internal/internaldns/model" + "go.miloapis.com/dns-operator/internal/internaldns/projectstatus" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// AckSink relies on NATS subject ACLs to bind each authenticated serving member +// to its own AckSubject. It also verifies membership, region, resource lifetime, +// revision and freshness before changing project-facing publication status. +type ProjectScope struct { + Client client.Client + Namespace string + RequestTimeout time.Duration +} + +type AckSink struct { + Client client.Client + Namespace string + ProjectNamespaces []string + Projects []ProjectScope + PublicationMembers []model.Member + PublicationRegions map[string][]model.Member + Region string + Shard string + Members []model.Member + Now func() time.Time +} + +func (s *AckSink) Apply(ctx context.Context, subject string, env model.Envelope) error { + if env.Kind != model.KindMemberAck || env.Region != s.Region || env.Shard != s.Shard { + return fmt.Errorf("ack belongs to another kind, region or shard") + } + var a model.MemberAck + if err := json.Unmarshal(env.Payload, &a); err != nil { + return err + } + if subject != model.AckSubject(s.Region, s.Shard, a.MemberID) || env.ResourceUID != a.ResourceUID || env.Epoch != a.Epoch || env.Revision != a.Revision { + return fmt.Errorf("ack subject or envelope identity mismatch") + } + var member *model.Member + for i := range s.Members { + if s.Members[i].MemberID == a.MemberID { + member = &s.Members[i] + break + } + } + if member == nil || member.ReplicaID != a.ReplicaID { + return fmt.Errorf("ack member identity is not assigned to this shard") + } + now := time.Now().UTC() + if s.Now != nil { + now = s.Now().UTC() + } + if a.ObservedAt.IsZero() || a.ObservedAt.After(now.Add(5*time.Second)) { + return fmt.Errorf("ack clock is outside the allowed bound") + } + if a.Phase == model.AckVerified { + if a.ValidUntil.IsZero() || a.ValidUntil.After(now.Add(2*time.Minute)) { + return fmt.Errorf("verified ack requires a bounded fresh serving lease") + } + // A durable replay may deliver a correctly authenticated health lease + // after its deadline. It can never make a member ready again, and + // retrying it would create a permanent redelivery loop. + if !a.ValidUntil.After(now) { + return nil + } + } + row := dnsv1alpha1.DNSApplyAcknowledgement{MemberID: a.MemberID, Phase: string(a.Phase), Revision: int64(a.Revision), ObservedAt: metav1.NewTime(a.ObservedAt), ObservedUnixNano: a.ObservedAt.UnixNano(), Error: a.Error} + if !a.ValidUntil.IsZero() { + until := metav1.NewTime(a.ValidUntil) + row.ValidUntil = &until + } + switch a.Kind { + case model.KindPublicationManifest: + if member.Role != dnsValueRegional && member.Role != dnsValueCluster { + return fmt.Errorf("publication ack is not from a regional resolver") + } + return s.publication(ctx, a, row, now) + case model.KindServingSnapshot: + if member.Role != "node" && member.Role != dnsValueCluster && member.Role != dnsValueResolver && member.Role != dnsValueRegional { + return fmt.Errorf("serving ack is not from a resolver member") + } + return s.serving(ctx, a, row, now) + default: + return fmt.Errorf("unsupported acknowledged kind") + } +} + +func (s *AckSink) publication(ctx context.Context, a model.MemberAck, row dnsv1alpha1.DNSApplyAcknowledgement, now time.Time) error { + var owner dnsv1alpha1.DNSPublicationOwnership + if err := s.Client.Get(ctx, client.ObjectKey{Namespace: s.Namespace, Name: "owner-" + model.OpaqueToken(a.ResourceUID)[:20]}, &owner); apierrors.IsNotFound(err) { + return nil + } else if err != nil { + return err + } + if string(owner.Spec.ZoneUID) != a.ResourceUID || owner.Spec.ActiveManifestName == "" { + return nil + } + var m dnsv1alpha1.DNSPublicationManifest + if err := s.Client.Get(ctx, client.ObjectKey{Namespace: s.Namespace, Name: owner.Spec.ActiveManifestName}, &m); err != nil { + return err + } + if uint64(m.Spec.WriterEpoch) != a.Epoch || uint64(m.Spec.Revision) != a.Revision { + return nil + } + row.WriterEpoch = int64(a.Epoch) + base := m.DeepCopy() + m.Status.ReplicaAcknowledgements = upsertAck(m.Status.ReplicaAcknowledgements, row) + complete := s.publicationVerified(&m, now) + condition(&m.Status.Conditions, "Published", complete, "ReplicaVerification", m.Generation, now) + if err := s.Client.Status().Patch(ctx, &m, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { + return err + } + // Project-facing status is projected by Refresh. Keeping source API calls + // out of this path lets the shared regional consumer durably acknowledge + // fleet health even while one source control plane is partitioned. + return nil +} + +func (s *AckSink) serving(ctx context.Context, a model.MemberAck, row dnsv1alpha1.DNSApplyAcknowledgement, now time.Time) error { + e, snapshot, err := LoadCommittedServingSnapshot(ctx, s.Client, s.Namespace, s.Region, s.Shard) + if err != nil { + return err + } + if e == nil || e.Epoch != a.Epoch || e.Revision != a.Revision { + return nil + } + var bindings dnsv1alpha1.DNSResolverBindingList + if err := s.Client.List(ctx, &bindings, client.InNamespace(s.Namespace)); err != nil { + return err + } + for _, b := range snapshot.Bindings { + if b.BindingUID != a.ResourceUID { + continue + } + for i := range bindings.Items { + binding := &bindings.Items[i] + if string(binding.UID) != b.BindingUID || binding.Spec.Configuration.Generation != int64(b.BindingGeneration) || binding.Spec.Configuration.Revision != int64(b.ConfigurationRevision) || binding.Spec.Authorization.Sequence != int64(b.Authorization.Revision) || binding.Spec.Authorization.WriterEpoch != int64(b.Authorization.IssuerEpoch) { + continue + } + base := binding.DeepCopy() + bindingRow := row + bindingRow.Revision = binding.Spec.Configuration.Revision + bindingRow.WriterEpoch = int64(a.Epoch) + bindingRow.SnapshotRevision = int64(a.Revision) + bindingRow.AuthorizationIssuerEpoch = binding.Spec.Authorization.WriterEpoch + bindingRow.AuthorizationRevision = binding.Spec.Authorization.Sequence + if bindingRow.ValidUntil != nil && bindingRow.ValidUntil.After(b.Authorization.ValidUntil) { + u := metav1.NewTime(b.Authorization.ValidUntil) + bindingRow.ValidUntil = &u + } + binding.Status.MemberAcknowledgements = upsertAck(binding.Status.MemberAcknowledgements, bindingRow) + complete := s.bindingVerified(binding, e, snapshot, now) + if complete { + binding.Status.Phase = dnsValueServing + binding.Status.ObservedConfigurationRevision = binding.Spec.Configuration.Revision + } else { + binding.Status.Phase = dnsValuePending + } + condition(&binding.Status.Conditions, dnsValueServing, complete, "MemberVerification", binding.Generation, now) + if err := s.Client.Status().Patch(ctx, binding, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { + return err + } + } + } + return nil +} + +func (s *AckSink) projectPublication(ctx context.Context, m *dnsv1alpha1.DNSPublicationManifest, complete bool, now time.Time) error { + projects := append([]ProjectScope(nil), s.Projects...) + for _, ns := range s.ProjectNamespaces { + projects = append(projects, ProjectScope{Client: s.Client, Namespace: ns}) + } + results := make(chan error, len(projects)) + for _, project := range projects { + + go func() { + projectCtx := ctx + cancel := func() {} + if project.RequestTimeout > 0 { + projectCtx, cancel = context.WithTimeout(ctx, project.RequestTimeout) + } + defer cancel() + results <- s.projectPublicationForScope(projectCtx, project, m, complete, now) + }() + } + var projectErrors error + for range projects { + projectErrors = errors.Join(projectErrors, <-results) + } + return projectErrors +} + +func (s *AckSink) projectPublicationForScope(ctx context.Context, project ProjectScope, m *dnsv1alpha1.DNSPublicationManifest, complete bool, now time.Time) error { + ns, projectClient := project.Namespace, project.Client + if projectClient == nil { + return fmt.Errorf("project %s status client unavailable", ns) + } + var projectErrors error + { + var zones dnsv1alpha1.DNSZoneList + if err := projectClient.List(ctx, &zones, client.InNamespace(ns)); err != nil { + return fmt.Errorf("project %s list zones: %w", ns, err) + } + for i := range zones.Items { + z := &zones.Items[i] + if z.UID != m.Spec.ZoneRef.UID { + continue + } + base := z.DeepCopy() + projectstatus.Published(&z.Status.Conditions, complete, z.Generation, now) + projectstatus.Programmed(&z.Status.Conditions, hasApplied(m.Status.ReplicaAcknowledgements, m.Spec.WriterEpoch, m.Spec.Revision), z.Generation, now) + if !jsonEqualStatus(base.Status, z.Status) { + if err := projectClient.Status().Patch(ctx, z, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { + projectErrors = errors.Join(projectErrors, fmt.Errorf("project %s update zone %s status: %w", ns, z.Name, err)) + } + } + var regs dnsv1alpha1.DNSRegistrationList + if err := projectClient.List(ctx, ®s, client.InNamespace(ns)); err != nil { + projectErrors = errors.Join(projectErrors, fmt.Errorf("project %s list registrations: %w", ns, err)) + continue + } + var chunks []model.PublicationChunk + for idx, ref := range m.Spec.Chunks { + var c dnsv1alpha1.DNSPublicationChunk + if err := s.Client.Get(ctx, client.ObjectKey{Namespace: s.Namespace, Name: ref.Name}, &c); err != nil { + return err + } + chunks = append(chunks, model.PublicationChunk{ManifestUID: m.Name, ZoneUID: string(m.Spec.ZoneRef.UID), WriterEpoch: uint64(m.Spec.WriterEpoch), Revision: uint64(m.Spec.Revision), Index: idx, SHA256: c.Spec.SHA256, Payload: c.Spec.Payload}) + } + manifest := model.PublicationManifest{ManifestUID: m.Name, ZoneUID: string(m.Spec.ZoneRef.UID), Apex: m.Spec.ZoneApex, WriterEpoch: uint64(m.Spec.WriterEpoch), Revision: uint64(m.Spec.Revision), Tombstone: m.Spec.Tombstone, ContentHash: m.Spec.ContentHash, GeneratedAt: m.Spec.GeneratedAt.Time} + for idx, ref := range m.Spec.Chunks { + manifest.Chunks = append(manifest.Chunks, model.ChunkRef{Index: idx, SHA256: ref.SHA256, Size: int(ref.Size)}) + } + payload, err := model.VerifyManifest(manifest, chunks) + if err != nil { + return err + } + var plan model.PublicationPlan + if len(payload) > 0 { + if err := json.Unmarshal(payload, &plan); err != nil { + return err + } + } + registrationGenerations := map[string]uint64{} + for _, fence := range plan.Registrations { + registrationGenerations[fence.UID] = fence.Generation + } + for j := range regs.Items { + r := ®s.Items[j] + if registrationGenerations[string(r.UID)] != uint64(r.Generation) { + continue + } + if r.Spec.DNSZoneRef.Name != z.Name || r.Spec.DNSZoneRef.UID != "" && r.Spec.DNSZoneRef.UID != z.UID { + continue + } + available := false + name := model.AbsoluteName(r.Spec.Name) + apex := model.AbsoluteName(z.Spec.DomainName) + if name == "@." { + name = apex + } else if !strings.HasSuffix(name, "."+apex) && name != apex { + name = strings.TrimSuffix(name, ".") + "." + apex + } + for _, rrset := range plan.RRSets { + if model.AbsoluteName(rrset.Name) != name { + continue + } + for _, record := range rrset.Records { + if record.Eligible && (record.ContributionUID == "" || record.ValidUntil.After(now)) { + available = true + } + } + } + base := r.DeepCopy() + r.Status.PublicationWriterEpoch = m.Spec.WriterEpoch + r.Status.PublicationRevision = m.Spec.Revision + projectstatus.Published(&r.Status.Conditions, complete, r.Generation, now) + projectstatus.Available(&r.Status.Conditions, available, r.Generation, now) + if !jsonEqualStatus(base.Status, r.Status) { + if err := projectClient.Status().Patch(ctx, r, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { + projectErrors = errors.Join(projectErrors, fmt.Errorf("project %s update registration %s status: %w", ns, r.Name, err)) + } + } + } + } + } + return projectErrors +} + +func (s *AckSink) verified(rows []dnsv1alpha1.DNSApplyAcknowledgement, now time.Time, role string, revision, epoch, authEpoch, authRevision int64) bool { + matches := func(member model.Member) bool { + for _, row := range rows { + if row.MemberID == member.MemberID && row.Revision == revision && row.WriterEpoch == epoch && (role != dnsValueResolver || row.AuthorizationIssuerEpoch == authEpoch && row.AuthorizationRevision == authRevision) && row.Phase == string(model.AckVerified) && row.ValidUntil != nil && row.ValidUntil.After(now) { + return true + } + } + return false + } + if role == "publication" { + groups := s.PublicationRegions + if len(groups) == 0 { + members := s.PublicationMembers + if len(members) == 0 { + members = s.Members + } + groups = map[string][]model.Member{"local": members} + } + for _, members := range groups { + required := 0 + verified := 0 + for _, member := range members { + if member.Role == dnsValueRegional || member.Role == dnsValueCluster { + required++ + if matches(member) { + verified++ + } + } + } + if required == 0 || verified != required { + return false + } + } + return len(groups) > 0 + } + required := 0 + for _, member := range s.Members { + match := member.Role == role + if role == dnsValueResolver { + match = member.Role == "node" || member.Role == dnsValueCluster || member.Role == dnsValueResolver || member.Role == dnsValueRegional + } + if !match { + continue + } + required++ + if !matches(member) { + return false + } + } + return required > 0 +} + +func upsertAck(rows []dnsv1alpha1.DNSApplyAcknowledgement, row dnsv1alpha1.DNSApplyAcknowledgement) []dnsv1alpha1.DNSApplyAcknowledgement { + for i := range rows { + if rows[i].MemberID == row.MemberID { + if row.ObservedUnixNano > 0 && rows[i].ObservedUnixNano > 0 && row.ObservedUnixNano < rows[i].ObservedUnixNano || row.ObservedAt.Before(&rows[i].ObservedAt) { + return rows + } + rows[i] = row + return rows + } + } + return append(rows, row) +} +func condition(rows *[]metav1.Condition, name string, value bool, reason string, generation int64, now time.Time) { + status := metav1.ConditionFalse + if value { + status = metav1.ConditionTrue + } + apimeta.SetStatusCondition(rows, metav1.Condition{Type: name, Status: status, Reason: reason, Message: reason, ObservedGeneration: generation, LastTransitionTime: metav1.NewTime(now)}) +} + +func hasApplied(rows []dnsv1alpha1.DNSApplyAcknowledgement, epoch, revision int64) bool { + for _, row := range rows { + if row.WriterEpoch == epoch && row.Revision == revision && (row.Phase == string(model.AckApplied) || row.Phase == string(model.AckVerified)) { + return true + } + } + return false +} + +// Refresh withdraws status readiness when a member lease or authorization expires, +// even if its process or event stream has stopped producing acknowledgements. +func (s *AckSink) Refresh(ctx context.Context) error { + now := time.Now().UTC() + if s.Now != nil { + now = s.Now().UTC() + } + var bindings dnsv1alpha1.DNSResolverBindingList + if err := s.Client.List(ctx, &bindings, client.InNamespace(s.Namespace), client.MatchingLabels{"internal-dns.miloapis.com/region": model.SafeToken(s.Region), "internal-dns.miloapis.com/shard": model.SafeToken(s.Shard)}); err != nil { + return err + } + envelope, snapshot, err := LoadCommittedServingSnapshot(ctx, s.Client, s.Namespace, s.Region, s.Shard) + if err != nil { + return err + } + for i := range bindings.Items { + b := &bindings.Items[i] + base := b.DeepCopy() + complete := s.bindingVerified(b, envelope, snapshot, now) + condition(&b.Status.Conditions, dnsValueServing, complete, "MemberVerification", b.Generation, now) + if complete { + b.Status.Phase = dnsValueServing + } else { + b.Status.Phase = dnsValuePending + } + if !jsonEqualStatus(base.Status, b.Status) { + if err := s.Client.Status().Patch(ctx, b, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { + return err + } + } + } + var owners dnsv1alpha1.DNSPublicationOwnershipList + if err := s.Client.List(ctx, &owners, client.InNamespace(s.Namespace)); err != nil { + return err + } + type projection struct { + manifest *dnsv1alpha1.DNSPublicationManifest + complete bool + } + projections := make([]projection, 0, len(owners.Items)) + var refreshErrors error + for _, owner := range owners.Items { + if owner.Spec.ActiveManifestName == "" { + continue + } + var m dnsv1alpha1.DNSPublicationManifest + if err := s.Client.Get(ctx, client.ObjectKey{Namespace: s.Namespace, Name: owner.Spec.ActiveManifestName}, &m); err != nil { + refreshErrors = errors.Join(refreshErrors, err) + continue + } + base := m.DeepCopy() + complete := s.publicationVerified(&m, now) + condition(&m.Status.Conditions, "Published", complete, "ReplicaVerification", m.Generation, now) + if !jsonEqualStatus(base.Status, m.Status) { + if err := s.Client.Status().Patch(ctx, &m, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { + refreshErrors = errors.Join(refreshErrors, err) + continue + } + } + manifest := m.DeepCopy() + projections = append(projections, projection{manifest: manifest, complete: complete}) + } + // Launch every manifest projection before waiting for any source. Thus one + // partitioned project cannot delay status for healthy projects or manifests. + results := make(chan error, len(projections)) + for _, p := range projections { + + go func() { results <- s.projectPublication(ctx, p.manifest, p.complete, now) }() + } + for range projections { + refreshErrors = errors.Join(refreshErrors, <-results) + } + return refreshErrors +} + +func jsonEqualStatus(a, b interface{}) bool { + x, _ := json.Marshal(a) + y, _ := json.Marshal(b) + return string(x) == string(y) +} + +func (s *AckSink) publicationVerified(m *dnsv1alpha1.DNSPublicationManifest, now time.Time) bool { + if len(m.Spec.ServingTargets) == 0 { + return m.Spec.Tombstone + } + scoped := *s + scoped.PublicationRegions = map[string][]model.Member{} + for _, target := range m.Spec.ServingTargets { + key := target.Region + "/" + target.Shard + members, ok := s.PublicationRegions[key] + if !ok && len(s.PublicationRegions) == 0 && target.Region == s.Region && target.Shard == s.Shard { + members = s.Members + ok = true + } + if !ok { + return false + } + scoped.PublicationRegions[key] = members + } + return scoped.verified(m.Status.ReplicaAcknowledgements, now, "publication", m.Spec.Revision, m.Spec.WriterEpoch, 0, 0) +} + +// LoadCommittedServingSnapshot reads the shard's durable commit pointer rather +// than accepting mutually consistent but retired member acknowledgements. +func LoadCommittedServingSnapshot(ctx context.Context, c client.Client, namespace, region, shard string) (*model.Envelope, *model.ServingSnapshot, error) { + var cm corev1.ConfigMap + if err := c.Get(ctx, client.ObjectKey{Namespace: namespace, Name: "dns-shard-owner-" + model.OpaqueToken(region+"/"+shard)}, &cm); apierrors.IsNotFound(err) { + return nil, nil, nil + } else if err != nil { + return nil, nil, err + } + var owner shardState + if err := json.Unmarshal([]byte(cm.Data["state"]), &owner); err != nil { + return nil, nil, err + } + if owner.ActiveOutbox == "" { + return nil, nil, nil + } + var out dnsv1alpha1.DNSTransportOutbox + if err := c.Get(ctx, client.ObjectKey{Namespace: namespace, Name: owner.ActiveOutbox}, &out); err != nil { + return nil, nil, err + } + if model.Hash(out.Spec.Payload) != out.Spec.PayloadHash { + return nil, nil, fmt.Errorf("committed serving snapshot hash mismatch") + } + var e model.Envelope + if err := json.Unmarshal(out.Spec.Payload, &e); err != nil { + return nil, nil, err + } + if e.Kind != model.KindServingSnapshot || e.Region != region || e.Shard != shard || e.Epoch != owner.Epoch || int64(e.Epoch) != out.Spec.WriterEpoch || int64(e.Revision) != out.Spec.Revision { + return nil, nil, fmt.Errorf("committed serving snapshot fence mismatch") + } + var snapshot model.ServingSnapshot + if err := json.Unmarshal(e.Payload, &snapshot); err != nil { + return nil, nil, err + } + if snapshot.Region != region || snapshot.Shard != shard || snapshot.ConfigurationEpoch != e.Epoch || snapshot.ConfigurationRevision != e.Revision { + return nil, nil, fmt.Errorf("committed serving snapshot body fence mismatch") + } + return &e, &snapshot, nil +} + +func (s *AckSink) bindingVerified(b *dnsv1alpha1.DNSResolverBinding, envelope *model.Envelope, snapshot *model.ServingSnapshot, now time.Time) bool { + if envelope == nil || snapshot == nil || b.Spec.Tombstone || !b.Spec.Authorization.ValidUntil.After(now) { + return false + } + present := false + for _, desired := range snapshot.Bindings { + if desired.BindingUID == string(b.UID) && !desired.Tombstone && int64(desired.BindingGeneration) == b.Spec.Configuration.Generation && int64(desired.ConfigurationRevision) == b.Spec.Configuration.Revision && int64(desired.Authorization.IssuerEpoch) == b.Spec.Authorization.WriterEpoch && int64(desired.Authorization.Revision) == b.Spec.Authorization.Sequence { + present = true + break + } + } + if !present { + return false + } + rows := make([]dnsv1alpha1.DNSApplyAcknowledgement, 0, len(b.Status.MemberAcknowledgements)) + for _, row := range b.Status.MemberAcknowledgements { + if row.SnapshotRevision == int64(envelope.Revision) { + rows = append(rows, row) + } + } + return s.verified(rows, now, dnsValueResolver, b.Spec.Configuration.Revision, int64(envelope.Epoch), b.Spec.Authorization.WriterEpoch, b.Spec.Authorization.Sequence) +} diff --git a/internal/internaldns/platform/outbox.go b/internal/internaldns/platform/outbox.go new file mode 100644 index 00000000..49767e78 --- /dev/null +++ b/internal/internaldns/platform/outbox.go @@ -0,0 +1,339 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package platform + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "time" + + dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" + "go.miloapis.com/dns-operator/internal/internaldns/model" + "go.miloapis.com/dns-operator/internal/internaldns/transport" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/fields" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +type BootstrapWriter interface { + Put(context.Context, string, model.Envelope) (uint64, error) +} + +// Outbox retries durable intent, including ambiguous publish outcomes, using a +// stable event ID. A NATS acknowledgement is transport durability, never serving +// readiness. Activation is fenced against the committed ownership pointer. +type Outbox struct { + Client client.Client + Namespace string + Region string + Publisher transport.ExportPublisher + Bootstrap BootstrapWriter + Now func() time.Time + // PendingOnly requires the CRD status.state selectable field (Kubernetes 1.32+). + PendingOnly bool +} + +func (o *Outbox) Step(ctx context.Context) error { + if o.Client == nil || o.Publisher == nil || o.Namespace == "" { + return fmt.Errorf("outbox client, publisher and namespace are required") + } + var list dnsv1alpha1.DNSTransportOutboxList + opts := []client.ListOption{client.InNamespace(o.Namespace)} + if o.Region != "" { + opts = append(opts, client.MatchingLabels{"internal-dns.miloapis.com/region": model.SafeToken(o.Region)}) + } + if o.PendingOnly { + opts = append(opts, client.MatchingFieldsSelector{Selector: fields.AndSelectors(fields.OneTermNotEqualSelector("status.state", dnsValueAcknowledged), fields.OneTermNotEqualSelector("status.state", dnsValueSuperseded))}) + } + if err := o.Client.List(ctx, &list, opts...); err != nil { + return err + } + sort.Slice(list.Items, func(i, j int) bool { + a, b := list.Items[i], list.Items[j] + if a.Spec.Activation != b.Spec.Activation { + return !a.Spec.Activation + } + return a.Name < b.Name + }) + states := map[string]string{} + for _, out := range list.Items { + states[out.Name] = out.Status.State + } + var first error + for i := range list.Items { + out := &list.Items[i] + if out.Status.State == dnsValueAcknowledged || out.Status.State == dnsValueSuperseded { + continue + } + var env model.Envelope + if model.Hash(out.Spec.Payload) != out.Spec.PayloadHash { + if first == nil { + first = fmt.Errorf("outbox %s payload hash mismatch", out.Name) + } + continue + } + if err := json.Unmarshal(out.Spec.Payload, &env); err != nil { + if first == nil { + first = err + } + continue + } + if err := env.Validate(); err != nil { + if first == nil { + first = err + } + continue + } + if env.ResourceUID != string(out.Spec.ResourceUID) || env.Epoch != uint64(out.Spec.WriterEpoch) || env.Revision != uint64(out.Spec.Revision) { + if first == nil { + first = fmt.Errorf("outbox envelope identity mismatch") + } + continue + } + if out.Spec.Subject != expectedSubject(env) { + if first == nil { + first = fmt.Errorf("outbox %s subject does not match its payload", out.Name) + } + continue + } + committed, err := o.committed(ctx, out, env) + if err != nil { + if first == nil { + first = err + } + continue + } + if !committed { + older, oldErr := o.superseded(ctx, out, env) + if oldErr != nil && first == nil { + first = oldErr + } + if older { + base := out.DeepCopy() + out.Status.State = dnsValueSuperseded + if err := o.Client.Status().Patch(ctx, out, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil && first == nil { + first = err + } + states[out.Name] = dnsValueSuperseded + } + continue + } + ready := true + for _, dep := range out.Spec.DependsOn { + ok, depErr := o.dependencyReady(ctx, env, dep, states) + if depErr != nil && first == nil { + first = depErr + } + if depErr != nil || !ok { + ready = false + break + } + } + if !ready { + continue + } + ack, err := o.Publisher.Publish(ctx, out.Spec.Subject, env) + if err == nil && o.Bootstrap != nil { + key, keyErr := model.SnapshotKey(env) + if keyErr != nil { + err = keyErr + } else { + _, err = o.Bootstrap.Put(ctx, key, env) + } + } + // Refetch before a status update to preserve concurrent acknowledgement + // changes and resourceVersion fencing; an old object UID cannot ACK a new one. + var current dnsv1alpha1.DNSTransportOutbox + if getErr := o.Client.Get(ctx, client.ObjectKeyFromObject(out), ¤t); getErr != nil { + if first == nil { + first = getErr + } + continue + } + if current.UID != out.UID || current.Spec.PayloadHash != out.Spec.PayloadHash { + if first == nil { + first = fmt.Errorf("outbox lifetime changed during publication") + } + continue + } + base := current.DeepCopy() + current.Status.Attempts++ + if err != nil { + current.Status.State = dnsValuePending + current.Status.LastError = err.Error() + if first == nil { + first = err + } + } else { + now := time.Now().UTC() + if o.Now != nil { + now = o.Now().UTC() + } + at := metav1.NewTime(now) + current.Status.State = dnsValueAcknowledged + current.Status.AcknowledgedAt = &at + current.Status.StreamSequence = ack.Sequence + current.Status.LastError = "" + states[out.Name] = dnsValueAcknowledged + } + if patchErr := o.Client.Status().Patch(ctx, ¤t, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); patchErr != nil { + if first == nil { + first = patchErr + } + continue + } + if err == nil && env.Kind == model.KindPublicationManifest { + var m dnsv1alpha1.DNSPublicationManifest + if getErr := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: out.Spec.ManifestRef.Name}, &m); getErr == nil { + base := m.DeepCopy() + m.Status.ExportState = dnsValueAcknowledged + m.Status.StreamSequence = ack.Sequence + m.Status.ExportedAt = current.Status.AcknowledgedAt + if patchErr := o.Client.Status().Patch(ctx, &m, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); patchErr != nil && first == nil { + first = patchErr + } + } + } + } + return first +} + +func (o *Outbox) committed(ctx context.Context, out *dnsv1alpha1.DNSTransportOutbox, env model.Envelope) (bool, error) { + if env.Kind == model.KindServingSnapshot { + name := out.Labels[shardOwnerLabel] + if name == "" { + return false, fmt.Errorf("serving outbox lacks its ownership record") + } + var cm corev1.ConfigMap + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: name}, &cm); err != nil { + return false, err + } + var s shardState + if err := json.Unmarshal([]byte(cm.Data["state"]), &s); err != nil { + return false, err + } + return cm.UID == out.Spec.ResourceUID && s.Epoch == env.Epoch && s.ActiveOutbox == out.Name, nil + } + if out.Spec.ManifestRef.Name == "" { + return false, fmt.Errorf("publication outbox lacks manifest reference") + } + var m dnsv1alpha1.DNSPublicationManifest + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: out.Spec.ManifestRef.Name}, &m); apierrors.IsNotFound(err) { + return false, nil + } else if err != nil { + return false, err + } + var own dnsv1alpha1.DNSPublicationOwnership + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: "owner-" + model.OpaqueToken(string(m.Spec.ZoneRef.UID))[:20]}, &own); apierrors.IsNotFound(err) { + return false, nil + } else if err != nil { + return false, err + } + return own.Spec.WriterEpoch == int64(env.Epoch) && own.Spec.ActiveManifestName == m.Name && m.Spec.WriterEpoch == int64(env.Epoch) && m.Spec.Revision == int64(env.Revision), nil +} + +func expectedSubject(env model.Envelope) string { + switch env.Kind { + case model.KindServingSnapshot: + return model.ServingSubject(env.Region, env.Shard) + case model.KindPublicationChunk: + return model.RecordChunkSubject(env.Region, env.Shard, env.ResourceUID) + case model.KindPublicationManifest: + return model.RecordManifestSubject(env.Region, env.Shard, env.ResourceUID) + default: + return "" + } +} + +// A serving snapshot pins a minimum publication fence, rather than a historical +// activation event. A newer committed complete manifest may satisfy it. Chunk +// dependencies within a manifest remain exact: revisions cannot be mixed. +func (o *Outbox) dependencyReady(ctx context.Context, env model.Envelope, name string, states map[string]string) (bool, error) { + if states[name] == dnsValueAcknowledged { + return true, nil + } + var dep dnsv1alpha1.DNSTransportOutbox + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: name}, &dep); apierrors.IsNotFound(err) { + return false, nil + } else if err != nil { + return false, err + } + if dep.Status.State == dnsValueAcknowledged { + states[name] = dep.Status.State + return true, nil + } + if env.Kind != model.KindServingSnapshot || dep.Spec.ManifestRef.Name == "" { + return false, nil + } + var previous dnsv1alpha1.DNSPublicationManifest + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: dep.Spec.ManifestRef.Name}, &previous); err != nil { + return false, err + } + var owner dnsv1alpha1.DNSPublicationOwnership + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: "owner-" + model.OpaqueToken(string(previous.Spec.ZoneRef.UID))[:20]}, &owner); err != nil { + return false, err + } + if owner.Spec.ActiveManifestName == "" { + return false, nil + } + var current dnsv1alpha1.DNSPublicationManifest + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: owner.Spec.ActiveManifestName}, ¤t); err != nil { + return false, err + } + if model.Compare(uint64(current.Spec.WriterEpoch), uint64(current.Spec.Revision), uint64(previous.Spec.WriterEpoch), uint64(previous.Spec.Revision)) < 0 { + return false, nil + } + var activation dnsv1alpha1.DNSTransportOutbox + key := client.ObjectKey{Namespace: o.Namespace, Name: model.PublicationActivationName(current.Name, env.Region, env.Shard)} + if err := o.Client.Get(ctx, key, &activation); apierrors.IsNotFound(err) { + return false, nil + } else if err != nil { + return false, err + } + return activation.Status.State == dnsValueAcknowledged, nil +} + +// An abandoned staged event remains pending until a strictly newer committed +// fence proves that it can never become active. This avoids unbounded pending +// scans without deleting the immutable audit/replay artifacts. +func (o *Outbox) superseded(ctx context.Context, out *dnsv1alpha1.DNSTransportOutbox, env model.Envelope) (bool, error) { + if env.Kind == model.KindServingSnapshot { + var cm corev1.ConfigMap + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: out.Labels[shardOwnerLabel]}, &cm); err != nil { + return false, err + } + var owner shardState + if err := json.Unmarshal([]byte(cm.Data["state"]), &owner); err != nil { + return false, err + } + if cm.UID != out.Spec.ResourceUID { + return true, nil + } + if owner.ActiveOutbox == "" { + return false, nil + } + var active dnsv1alpha1.DNSTransportOutbox + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: owner.ActiveOutbox}, &active); err != nil { + return false, err + } + return model.Compare(uint64(active.Spec.WriterEpoch), uint64(active.Spec.Revision), env.Epoch, env.Revision) > 0, nil + } + var owner dnsv1alpha1.DNSPublicationOwnership + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: "owner-" + model.OpaqueToken(env.ResourceUID)[:20]}, &owner); apierrors.IsNotFound(err) { + return false, nil + } else if err != nil { + return false, err + } + if owner.Spec.ActiveManifestName == "" { + return false, nil + } + var active dnsv1alpha1.DNSPublicationManifest + if err := o.Client.Get(ctx, client.ObjectKey{Namespace: o.Namespace, Name: owner.Spec.ActiveManifestName}, &active); err != nil { + return false, err + } + return model.Compare(uint64(active.Spec.WriterEpoch), uint64(active.Spec.Revision), env.Epoch, env.Revision) > 0, nil +} diff --git a/internal/internaldns/platform/planner.go b/internal/internaldns/platform/planner.go new file mode 100644 index 00000000..1c7ddd2f --- /dev/null +++ b/internal/internaldns/platform/planner.go @@ -0,0 +1,273 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package platform + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + "sort" + "time" + + dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" + "go.miloapis.com/dns-operator/internal/internaldns/model" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const shardOwnerLabel = "internal-dns.miloapis.com/shard-owner" + +type PlannerConfig struct { + Namespace string + Region string + Shard string + Identity string + LeaseDuration time.Duration + NodeBackends []model.Backend + ClusterBackends []model.Backend + Members []model.Member + // Hard safety budgets; production values require capacity qualification. + MaxBindings int `json:"maxBindings,omitempty"` + MaxSnapshotBytes int `json:"maxSnapshotBytes,omitempty"` +} + +type shardState struct { + Holder string `json:"holder"` + Epoch uint64 `json:"epoch"` + NextRevision uint64 `json:"nextRevision"` + LeaseUntil time.Time `json:"leaseUntil"` + ActiveOutbox string `json:"activeOutbox,omitempty"` + ContentHash string `json:"contentHash,omitempty"` +} + +// Planner is the single fenced logical writer of a complete shard snapshot. +// Project reconcilers only create binding data; this planner merges every project +// in the shared shard so a tenant's update cannot erase another tenant's views. +type Planner struct { + Client client.Client + Allocator *Allocator + Config PlannerConfig + Now func() time.Time +} + +func (p *Planner) Step(ctx context.Context) error { + c := p.Config + if p.Client == nil || p.Allocator == nil || c.Namespace == "" || c.Region == "" || c.Shard == "" || c.Identity == "" { + return fmt.Errorf("shared planner requires client, allocator, namespace, region, shard and holder identity") + } + now := time.Now().UTC() + if p.Now != nil { + now = p.Now().UTC() + } + lease := c.LeaseDuration + if lease <= 0 { + lease = 30 * time.Second + } + key := client.ObjectKey{Namespace: c.Namespace, Name: "dns-shard-owner-" + model.OpaqueToken(c.Region+"/"+c.Shard)} + var owner corev1.ConfigMap + if err := p.Client.Get(ctx, key, &owner); apierrors.IsNotFound(err) { + state := shardState{Holder: c.Identity, Epoch: 1, NextRevision: 1, LeaseUntil: now.Add(lease)} + b, _ := json.Marshal(state) + owner = corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: key.Namespace}, Data: map[string]string{"state": string(b)}} + if err := p.Client.Create(ctx, &owner); err != nil && !apierrors.IsAlreadyExists(err) { + return err + } + return nil + } else if err != nil { + return err + } + var state shardState + if err := json.Unmarshal([]byte(owner.Data["state"]), &state); err != nil { + return fmt.Errorf("invalid shard ownership: %w", err) + } + if state.Holder != c.Identity { + if state.LeaseUntil.After(now) { + return nil + } + state.Holder = c.Identity + state.Epoch++ + state.ContentHash = "" + state.LeaseUntil = now.Add(lease) + b, _ := json.Marshal(state) + owner.Data["state"] = string(b) + if err := p.Client.Update(ctx, &owner); err != nil { + return err + } + } + var bindings dnsv1alpha1.DNSResolverBindingList + if err := p.Client.List(ctx, &bindings, client.InNamespace(c.Namespace), client.MatchingLabels{"internal-dns.miloapis.com/region": model.SafeToken(c.Region), "internal-dns.miloapis.com/shard": model.SafeToken(c.Shard)}); err != nil { + return err + } + byZone := map[string]dnsv1alpha1.DNSPublicationOwnership{} + for _, binding := range bindings.Items { + for _, zoneRef := range binding.Spec.Configuration.ZoneRefs { + uid := zoneRef.UID + if _, exists := byZone[string(uid)]; exists { + continue + } + var owner dnsv1alpha1.DNSPublicationOwnership + if err := p.Client.Get(ctx, client.ObjectKey{Namespace: c.Namespace, Name: "owner-" + model.OpaqueToken(string(uid))[:20]}, &owner); apierrors.IsNotFound(err) { + continue + } else if err != nil { + return err + } + byZone[string(uid)] = owner + } + } + snapshot := model.ServingSnapshot{Region: c.Region, Shard: c.Shard, ConfigurationEpoch: state.Epoch, ConfigurationRevision: state.NextRevision, GeneratedAt: now, Members: append([]model.Member(nil), c.Members...)} + var prior model.ServingSnapshot + if state.ActiveOutbox != "" { + var out dnsv1alpha1.DNSTransportOutbox + if err := p.Client.Get(ctx, client.ObjectKey{Namespace: c.Namespace, Name: state.ActiveOutbox}, &out); err != nil { + return err + } + var e model.Envelope + if err := json.Unmarshal(out.Spec.Payload, &e); err != nil { + return err + } + if err := json.Unmarshal(e.Payload, &prior); err != nil { + return err + } + } + priorBindings := map[string]model.Binding{} + for _, b := range prior.Bindings { + priorBindings[b.BindingUID] = b + } + dependencies := map[string]bool{} + for _, obj := range bindings.Items { + b := obj.Spec + if b.Placement.Region != c.Region || b.Placement.Shard != c.Shard || b.Tombstone || !obj.DeletionTimestamp.IsZero() { + continue + } + if b.Configuration.Listeners.Node.Port != 53 || b.Configuration.Listeners.Regional.Port != 53 || !reflect.DeepEqual(b.Configuration.Listeners.Node.Transports, b.Configuration.Listeners.Regional.Transports) { + return fmt.Errorf("binding %s requires matching DNS listener ports and transports", obj.Name) + } + view := model.Binding{BindingUID: string(obj.UID), ProjectUID: string(b.Source.ProjectUID), ContextUID: string(b.Source.ResolverContextRef.UID), BindingGeneration: uint64(b.Configuration.Generation), ConfigurationRevision: uint64(b.Configuration.Revision), ConsumerAddress: b.Configuration.Listeners.Node.Address, ClusterAddress: b.Configuration.Listeners.Regional.Address, Port: uint16(b.Configuration.Listeners.Node.Port), NodeBackends: c.NodeBackends, ClusterBackends: c.ClusterBackends, Authorization: model.Authorization{IssuerEpoch: uint64(b.Authorization.WriterEpoch), Revision: uint64(b.Authorization.Sequence), ValidUntil: b.Authorization.ValidUntil.Time}} + for _, t := range b.Configuration.Listeners.Node.Transports { + view.Transports = append(view.Transports, model.Transport(t)) + } + complete := true + for _, zoneRef := range b.Configuration.ZoneRefs { + uid := zoneRef.UID + o, ok := byZone[string(uid)] + if !ok || o.Spec.ActiveManifestName == "" { + complete = false + break + } + var m dnsv1alpha1.DNSPublicationManifest + if err := p.Client.Get(ctx, client.ObjectKey{Namespace: c.Namespace, Name: o.Spec.ActiveManifestName}, &m); err != nil { + return err + } + targeted := false + for _, target := range m.Spec.ServingTargets { + if target.Region == c.Region && target.Shard == c.Shard { + targeted = true + break + } + } + if m.Spec.ZoneRef.UID != uid || m.Spec.Tombstone || !targeted { + complete = false + break + } + attachment := model.ZoneAttachment{ZoneUID: string(uid), Apex: m.Spec.ZoneApex, RequiredPublicationEpoch: uint64(m.Spec.WriterEpoch), RequiredPublicationRevision: uint64(m.Spec.Revision)} + // Record updates do not change a view's minimum activation dependency. + if old, ok := priorBindings[view.BindingUID]; ok && old.BindingGeneration == view.BindingGeneration { + for _, z := range old.Zones { + if z.ZoneUID == attachment.ZoneUID && z.Apex == attachment.Apex { + attachment.RequiredPublicationEpoch = z.RequiredPublicationEpoch + attachment.RequiredPublicationRevision = z.RequiredPublicationRevision + } + } + } + activationName := model.PublicationActivationName(m.Name, c.Region, c.Shard) + view.Zones = append(view.Zones, attachment) + dependencies[activationName] = true + } + if !complete { + continue + } + sort.Slice(view.Zones, func(i, j int) bool { return view.Zones[i].ZoneUID < view.Zones[j].ZoneUID }) + snapshot.Bindings = append(snapshot.Bindings, view) + } + sort.Slice(snapshot.Bindings, func(i, j int) bool { return snapshot.Bindings[i].BindingUID < snapshot.Bindings[j].BindingUID }) + sort.Slice(snapshot.Members, func(i, j int) bool { return snapshot.Members[i].MemberID < snapshot.Members[j].MemberID }) + maxBindings := c.MaxBindings + if maxBindings <= 0 { + maxBindings = 256 + } + if len(snapshot.Bindings) > maxBindings { + return fmt.Errorf("shard %s/%s exceeds binding budget: %d > %d; no partial snapshot will be exported", c.Region, c.Shard, len(snapshot.Bindings), maxBindings) + } + if err := snapshot.Validate(); err != nil { + return err + } + comparison := snapshot + comparison.GeneratedAt = time.Time{} + comparison.ConfigurationEpoch = 0 + comparison.ConfigurationRevision = 0 + b, _ := json.Marshal(comparison) + hash := model.Hash(b) + if state.ContentHash == hash { + if state.LeaseUntil.Sub(now) > lease/2 { + return nil + } + state.LeaseUntil = now.Add(lease) + b, _ := json.Marshal(state) + owner.Data["state"] = string(b) + return p.Client.Update(ctx, &owner) + } + if !state.LeaseUntil.After(now) { + return fmt.Errorf("shared planner lost its lease while staging snapshot") + } + // Reserve the revision before writing immutable artifacts. A crash after + // staging leaves an orphan, and retry uses a fresh revision instead of trying + // to rewrite an immutable event with a different timestamp or desired state. + revision := state.NextRevision + state.NextRevision++ + state.LeaseUntil = now.Add(lease) + reserved, _ := json.Marshal(state) + owner.Data["state"] = string(reserved) + if err := p.Client.Update(ctx, &owner); err != nil { + return err + } + name := fmt.Sprintf("dns-serving-%s-e%d-r%d", model.OpaqueToken(c.Region+"/"+c.Shard), state.Epoch, revision) + env, err := model.NewEnvelope(model.KindServingSnapshot, name, c.Region, c.Shard, string(owner.UID), state.Epoch, revision, now, snapshot) + if err != nil { + return err + } + payload, _ := json.Marshal(env) + maxBytes := c.MaxSnapshotBytes + if maxBytes <= 0 { + maxBytes = 512 * 1024 + } + if len(payload) > maxBytes { + return fmt.Errorf("shard serving envelope exceeds configured transport budget: %d > %d", len(payload), maxBytes) + } + deps := make([]string, 0, len(dependencies)) + for name := range dependencies { + deps = append(deps, name) + } + sort.Strings(deps) + out := &dnsv1alpha1.DNSTransportOutbox{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: c.Namespace, Labels: map[string]string{shardOwnerLabel: owner.Name, "internal-dns.miloapis.com/region": model.SafeToken(c.Region), "internal-dns.miloapis.com/shard": model.SafeToken(c.Shard)}}, Spec: dnsv1alpha1.DNSTransportOutboxSpec{Subject: model.ServingSubject(c.Region, c.Shard), ResourceUID: owner.UID, WriterEpoch: int64(state.Epoch), Revision: int64(revision), PayloadHash: model.Hash(payload), Payload: payload, Activation: true, DependsOn: deps}} + if err := p.Client.Create(ctx, out); apierrors.IsAlreadyExists(err) { + var existing dnsv1alpha1.DNSTransportOutbox + if err := p.Client.Get(ctx, client.ObjectKeyFromObject(out), &existing); err != nil { + return err + } + if !reflect.DeepEqual(existing.Spec, out.Spec) { + return fmt.Errorf("immutable serving revision collision") + } + } else if err != nil { + return err + } + state.ActiveOutbox = name + state.ContentHash = hash + state.LeaseUntil = now.Add(lease) + b, _ = json.Marshal(state) + owner.Data["state"] = string(b) + // Publication is enabled only after this resourceVersion-checked commit. + return p.Client.Update(ctx, &owner) +} diff --git a/internal/internaldns/platform/platform_test.go b/internal/internaldns/platform/platform_test.go new file mode 100644 index 00000000..52221e90 --- /dev/null +++ b/internal/internaldns/platform/platform_test.go @@ -0,0 +1,457 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package platform + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "testing" + "time" + + dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" + "go.miloapis.com/dns-operator/internal/internaldns/model" + "go.miloapis.com/dns-operator/internal/internaldns/transport" + corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +type blockingListClient struct { + client.Client + started chan struct{} + once sync.Once +} + +func (c *blockingListClient) List(ctx context.Context, _ client.ObjectList, _ ...client.ListOption) error { + c.once.Do(func() { close(c.started) }) + <-ctx.Done() + return ctx.Err() +} + +func TestProjectPublicationContinuesAfterSourceFailure(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 10, 6, 18, 0, 0, 0, time.UTC) + plan := model.PublicationPlan{ZoneUID: "zone-healthy", Apex: "healthy.internal", Registrations: []model.RegistrationFence{{UID: "registration-healthy", Generation: 2}}, RRSets: []model.RRSet{}} + payload, err := json.Marshal(plan) + if err != nil { + t.Fatal(err) + } + chunk := &dnsv1alpha1.DNSPublicationChunk{ObjectMeta: metav1.ObjectMeta{Name: "chunk", Namespace: "platform"}, Spec: dnsv1alpha1.DNSPublicationChunkSpec{WriterEpoch: 1, Revision: 3, Index: 0, SHA256: model.Hash(payload), Payload: payload}} + manifest := &dnsv1alpha1.DNSPublicationManifest{ObjectMeta: metav1.ObjectMeta{Name: "manifest", Namespace: "platform"}, Spec: dnsv1alpha1.DNSPublicationManifestSpec{ + ZoneRef: dnsv1alpha1.DNSObjectReference{UID: "zone-healthy"}, ZoneApex: "healthy.internal", WriterEpoch: 1, Revision: 3, + Chunks: []dnsv1alpha1.DNSPublicationChunkReference{{Name: chunk.Name, SHA256: chunk.Spec.SHA256, Size: int32(len(payload))}}, ContentHash: model.Hash(payload), GeneratedAt: metav1.NewTime(now), + }} + scheme := runtime.NewScheme() + if err := dnsv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + zone := &dnsv1alpha1.DNSZone{ObjectMeta: metav1.ObjectMeta{Name: "healthy", Namespace: "healthy-project", UID: "zone-healthy", Generation: 4}, Spec: dnsv1alpha1.DNSZoneSpec{DomainName: "healthy.internal", Visibility: dnsv1alpha1.DNSZoneVisibilityPrivate}} + registration := &dnsv1alpha1.DNSRegistration{ObjectMeta: metav1.ObjectMeta{Name: "api", Namespace: zone.Namespace, UID: "registration-healthy", Generation: 2}, Spec: dnsv1alpha1.DNSRegistrationSpec{DNSZoneRef: dnsv1alpha1.DNSObjectReference{Name: zone.Name, UID: zone.UID}, Name: "api"}} + healthy := fake.NewClientBuilder().WithScheme(scheme).WithObjects(zone, registration).WithStatusSubresource(&dnsv1alpha1.DNSZone{}, &dnsv1alpha1.DNSRegistration{}).Build() + blocked := &blockingListClient{Client: healthy, started: make(chan struct{})} + sink := AckSink{Client: testClient(t, chunk), Namespace: "platform", Projects: []ProjectScope{{Client: blocked, Namespace: "offline-project", RequestTimeout: 200 * time.Millisecond}, {Client: healthy, Namespace: zone.Namespace}}} + projectionDone := make(chan error, 1) + go func() { projectionDone <- sink.projectPublication(ctx, manifest, true, now) }() + <-blocked.started + deadline := time.Now().Add(150 * time.Millisecond) + var updated dnsv1alpha1.DNSZone + for { + if err := healthy.Get(ctx, client.ObjectKeyFromObject(zone), &updated); err != nil { + t.Fatal(err) + } + condition := apimeta.FindStatusCondition(updated.Status.Conditions, "Published") + if condition != nil && condition.Status == metav1.ConditionTrue { + break + } + if time.Now().After(deadline) { + t.Fatal("healthy source status waited for the partitioned source timeout") + } + time.Sleep(time.Millisecond) + } + if err := <-projectionDone; err == nil { + t.Fatal("offline source timeout was not reported") + } + condition := apimeta.FindStatusCondition(updated.Status.Conditions, "Published") + if condition == nil || condition.Status != metav1.ConditionTrue { + t.Fatalf("healthy source status was blocked by offline project: %#v", updated.Status.Conditions) + } + zoneResourceVersion := updated.ResourceVersion + var updatedRegistration dnsv1alpha1.DNSRegistration + if err := healthy.Get(ctx, client.ObjectKeyFromObject(registration), &updatedRegistration); err != nil { + t.Fatal(err) + } + registrationResourceVersion := updatedRegistration.ResourceVersion + if err := sink.projectPublication(ctx, manifest, true, now.Add(time.Minute)); err == nil { + t.Fatal("offline source failure was not reported on repeated projection") + } + if err := healthy.Get(ctx, client.ObjectKeyFromObject(zone), &updated); err != nil { + t.Fatal(err) + } + if err := healthy.Get(ctx, client.ObjectKeyFromObject(registration), &updatedRegistration); err != nil { + t.Fatal(err) + } + if updated.ResourceVersion != zoneResourceVersion || updatedRegistration.ResourceVersion != registrationResourceVersion { + t.Fatalf("converged project status churned: zone %s→%s registration %s→%s", zoneResourceVersion, updated.ResourceVersion, registrationResourceVersion, updatedRegistration.ResourceVersion) + } +} + +func TestSharedPlannerPreservesOtherProjects(t *testing.T) { + now := time.Date(2026, 10, 6, 18, 0, 0, 0, time.UTC) + region, shard, ns := "east", "shared-01", "platform" + state := shardState{Holder: "planner", Epoch: 1, NextRevision: 1, LeaseUntil: now.Add(time.Minute)} + b, _ := json.Marshal(state) + owner := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "dns-shard-owner-" + model.OpaqueToken(region+"/"+shard), Namespace: ns, UID: "shard-owner-uid"}, Data: map[string]string{"state": string(b)}} + objects := []client.Object{owner} + for i := 1; i <= 2; i++ { + zuid := types.UID(fmt.Sprintf("zone-%d", i)) + vuid := types.UID(fmt.Sprintf("vpc-%d", i)) + mname := fmt.Sprintf("manifest-%d", i) + objects = append(objects, &dnsv1alpha1.DNSPublicationOwnership{ObjectMeta: metav1.ObjectMeta{Name: "owner-" + model.OpaqueToken(string(zuid))[:20], Namespace: ns}, Spec: dnsv1alpha1.DNSPublicationOwnershipSpec{ZoneUID: zuid, WriterEpoch: 1, ActiveManifestName: mname}}) + objects = append(objects, &dnsv1alpha1.DNSPublicationManifest{ObjectMeta: metav1.ObjectMeta{Name: mname, Namespace: ns}, Spec: dnsv1alpha1.DNSPublicationManifestSpec{ZoneRef: dnsv1alpha1.DNSObjectReference{UID: zuid}, ZoneApex: "prod.internal", WriterEpoch: 1, Revision: 1, ServingTargets: []dnsv1alpha1.DNSPublicationTarget{{Region: region, Shard: shard}}}}) + objects = append(objects, &dnsv1alpha1.DNSResolverBinding{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("binding-%d", i), Namespace: ns, UID: types.UID(fmt.Sprintf("binding-uid-%d", i)), Labels: map[string]string{"internal-dns.miloapis.com/region": region, "internal-dns.miloapis.com/shard": shard}}, Spec: dnsv1alpha1.DNSResolverBindingSpec{Source: dnsv1alpha1.DNSResolverBindingSource{ProjectUID: types.UID(fmt.Sprintf("project-%d", i)), ResolverContextRef: dnsv1alpha1.DNSObjectReference{Name: string(vuid), UID: vuid}}, Placement: dnsv1alpha1.DNSResolverBindingPlacement{Region: region, Shard: shard}, Configuration: dnsv1alpha1.DNSResolverBindingConfiguration{Generation: 1, Revision: 1, Listeners: dnsv1alpha1.DNSResolverBindingListeners{Node: dnsv1alpha1.DNSResolverBindingListener{Address: fmt.Sprintf("fd53::%d", i), Port: 53, Transports: []string{"UDP", "TCP"}}, Regional: dnsv1alpha1.DNSResolverBindingListener{Address: fmt.Sprintf("fd54::%d", i), Port: 53, Transports: []string{"UDP", "TCP"}}}, ZoneRefs: zoneReferences([]types.UID{zuid})}, Authorization: dnsv1alpha1.DNSResolverAccessAuthorization{WriterEpoch: 1, Sequence: 1, ValidUntil: metav1.NewTime(now.Add(time.Minute))}}}) + } + cl := testClient(t, objects...) + p := Planner{Client: cl, Allocator: &Allocator{Client: cl, Namespace: ns}, Config: PlannerConfig{Namespace: ns, Region: region, Shard: shard, Identity: "planner", ClusterBackends: []model.Backend{{MemberID: "cluster-1", Address: "fd00::20", Port: 5300}}}, Now: func() time.Time { return now }} + if err := p.Step(context.Background()); err != nil { + t.Fatal(err) + } + var out dnsv1alpha1.DNSTransportOutboxList + if err := cl.List(context.Background(), &out); err != nil { + t.Fatal(err) + } + if len(out.Items) != 1 { + t.Fatalf("expected one shared snapshot, got %d", len(out.Items)) + } + var env model.Envelope + var snapshot model.ServingSnapshot + _ = json.Unmarshal(out.Items[0].Spec.Payload, &env) + _ = json.Unmarshal(env.Payload, &snapshot) + if len(snapshot.Bindings) != 2 { + t.Fatal("one project's update removed another project") + } + if err := p.Step(context.Background()); err != nil { + t.Fatal(err) + } + _ = cl.List(context.Background(), &out) + if len(out.Items) != 1 { + t.Fatal("unchanged desired state generated a new configuration revision") + } + var untargeted dnsv1alpha1.DNSPublicationManifest + if err := cl.Get(context.Background(), client.ObjectKey{Namespace: ns, Name: "manifest-1"}, &untargeted); err != nil { + t.Fatal(err) + } + untargeted.Spec.ServingTargets = nil + if err := cl.Update(context.Background(), &untargeted); err != nil { + t.Fatal(err) + } + if err := p.Step(context.Background()); err != nil { + t.Fatal(err) + } + _ = cl.List(context.Background(), &out) + if len(out.Items) != 2 { + t.Fatal("regional export removal did not change the shared snapshot") + } + var updatedOwner corev1.ConfigMap + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(owner), &updatedOwner); err != nil { + t.Fatal(err) + } + var updatedState shardState + _ = json.Unmarshal([]byte(updatedOwner.Data["state"]), &updatedState) + var current dnsv1alpha1.DNSTransportOutbox + if err := cl.Get(context.Background(), client.ObjectKey{Namespace: ns, Name: updatedState.ActiveOutbox}, ¤t); err != nil { + t.Fatal(err) + } + _ = json.Unmarshal(current.Spec.Payload, &env) + _ = json.Unmarshal(env.Payload, &snapshot) + if len(snapshot.Bindings) != 1 || snapshot.Bindings[0].ContextUID != "vpc-2" { + t.Fatal("untargeted zone entered snapshot or removed another project") + } + p.Config.Identity = "retired-other-owner" + if err := p.Step(context.Background()); err != nil { + t.Fatal(err) + } + _ = cl.List(context.Background(), &out) + if len(out.Items) != 2 { + t.Fatal("another holder published during a current lease") + } +} + +type recordingPublisher struct{ events []model.Envelope } + +func (p *recordingPublisher) Publish(_ context.Context, _ string, e model.Envelope) (transport.PublishAck, error) { + p.events = append(p.events, e) + return transport.PublishAck{Sequence: uint64(len(p.events))}, nil +} + +func TestOutboxDoesNotPublishAnUncommittedRevision(t *testing.T) { + now := time.Now().UTC() + state := shardState{Holder: "planner", Epoch: 1, NextRevision: 2, LeaseUntil: now.Add(time.Minute)} + b, _ := json.Marshal(state) + owner := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "shard-owner", Namespace: "platform", UID: "owner-uid"}, Data: map[string]string{"state": string(b)}} + snapshot := model.ServingSnapshot{Region: "east", Shard: "shared", ConfigurationEpoch: 1, ConfigurationRevision: 1, GeneratedAt: now} + env, err := model.NewEnvelope(model.KindServingSnapshot, "event-1", "east", "shared", "owner-uid", 1, 1, now, snapshot) + if err != nil { + t.Fatal(err) + } + payload, _ := json.Marshal(env) + out := &dnsv1alpha1.DNSTransportOutbox{ObjectMeta: metav1.ObjectMeta{Name: "event-1", Namespace: "platform", UID: "outbox-uid", Labels: map[string]string{shardOwnerLabel: owner.Name}}, Spec: dnsv1alpha1.DNSTransportOutboxSpec{Subject: model.ServingSubject("east", "shared"), ResourceUID: owner.UID, WriterEpoch: 1, Revision: 1, Activation: true, Payload: payload, PayloadHash: model.Hash(payload)}} + cl := testClient(t, owner, out) + pub := &recordingPublisher{} + export := Outbox{Client: cl, Namespace: "platform", Publisher: pub} + if err := export.Step(context.Background()); err != nil { + t.Fatal(err) + } + if len(pub.events) != 0 { + t.Fatal("uncommitted staged revision escaped through NATS") + } + var current corev1.ConfigMap + _ = cl.Get(context.Background(), client.ObjectKeyFromObject(owner), ¤t) + state.ActiveOutbox = out.Name + b, _ = json.Marshal(state) + current.Data["state"] = string(b) + if err := cl.Update(context.Background(), ¤t); err != nil { + t.Fatal(err) + } + if err := export.Step(context.Background()); err != nil { + t.Fatal(err) + } + if len(pub.events) != 1 { + t.Fatal("committed revision was not exported") + } + if err := export.Step(context.Background()); err != nil { + t.Fatal(err) + } + if len(pub.events) != 1 { + t.Fatal("durable acknowledged outbox was exported again") + } +} + +func TestAckCannotForgeAnotherMemberSubject(t *testing.T) { + now := time.Now().UTC() + a := model.MemberAck{MemberID: "auth-a", ReplicaID: "replica-a", ResourceUID: "zone-uid", Kind: model.KindPublicationManifest, Epoch: 1, Revision: 1, Phase: model.AckVerified, ObservedAt: now, ValidUntil: now.Add(30 * time.Second)} + env, err := model.NewEnvelope(model.KindMemberAck, "ack-1", "east", "shared", "zone-uid", 1, 1, now, a) + if err != nil { + t.Fatal(err) + } + sink := AckSink{Client: testClient(t), Namespace: "platform", Region: "east", Shard: "shared", Members: []model.Member{{MemberID: "auth-a", ReplicaID: "replica-a", Role: "regional"}}} + if err := sink.Apply(context.Background(), model.AckSubject("east", "shared", "auth-b"), env); err == nil { + t.Fatal("forged member acknowledgement was accepted") + } + if err := sink.Apply(context.Background(), model.AckSubject("east", "shared", "auth-a"), env); err != nil { + t.Fatal(err) + } +} + +func TestPublicationAckDoesNotWaitForSourceProjection(t *testing.T) { + now := time.Now().UTC() + zoneUID := types.UID("zone-uid") + manifest := &dnsv1alpha1.DNSPublicationManifest{ObjectMeta: metav1.ObjectMeta{Name: "manifest", Namespace: "platform"}, Spec: dnsv1alpha1.DNSPublicationManifestSpec{ + ZoneRef: dnsv1alpha1.DNSObjectReference{UID: zoneUID}, WriterEpoch: 1, Revision: 1, + ServingTargets: []dnsv1alpha1.DNSPublicationTarget{{Region: "east", Shard: "shared"}}, + }} + owner := &dnsv1alpha1.DNSPublicationOwnership{ObjectMeta: metav1.ObjectMeta{Name: "owner-" + model.OpaqueToken(string(zoneUID))[:20], Namespace: "platform"}, Spec: dnsv1alpha1.DNSPublicationOwnershipSpec{ZoneUID: zoneUID, WriterEpoch: 1, ActiveManifestName: manifest.Name}} + platformClient := testClient(t, owner, manifest) + blocked := &blockingListClient{Client: platformClient, started: make(chan struct{})} + sink := AckSink{Client: platformClient, Namespace: "platform", Region: "east", Shard: "shared", + Projects: []ProjectScope{{Client: blocked, Namespace: "offline", RequestTimeout: time.Second}}, + Members: []model.Member{{MemberID: "auth", ReplicaID: "replica", Role: "regional"}}, + PublicationRegions: map[string][]model.Member{ + "east/shared": {{MemberID: "auth", ReplicaID: "replica", Role: "regional"}}, + }, + Now: func() time.Time { return now }, + } + ack := model.MemberAck{MemberID: "auth", ReplicaID: "replica", ResourceUID: string(zoneUID), Kind: model.KindPublicationManifest, Epoch: 1, Revision: 1, Phase: model.AckVerified, ObservedAt: now, ValidUntil: now.Add(30 * time.Second)} + env, err := model.NewEnvelope(model.KindMemberAck, "ack", "east", "shared", string(zoneUID), 1, 1, now, ack) + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- sink.Apply(context.Background(), model.AckSubject("east", "shared", "auth"), env) }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(100 * time.Millisecond): + t.Fatal("platform acknowledgement waited for a partitioned source API") + } + select { + case <-blocked.started: + t.Fatal("ACK path attempted source status projection") + default: + } + var updated dnsv1alpha1.DNSPublicationManifest + if err := platformClient.Get(context.Background(), client.ObjectKeyFromObject(manifest), &updated); err != nil { + t.Fatal(err) + } + if len(updated.Status.ReplicaAcknowledgements) != 1 { + t.Fatalf("platform ACK was not persisted: %#v", updated.Status) + } +} + +func TestReadinessCannotMixAuthorizationRenewals(t *testing.T) { + now := time.Now().UTC() + until := metav1.NewTime(now.Add(time.Minute)) + s := AckSink{Members: []model.Member{{MemberID: "node", Role: "node"}, {MemberID: "cluster", Role: "cluster"}}} + rows := []dnsv1alpha1.DNSApplyAcknowledgement{ + {MemberID: "node", WriterEpoch: 2, Revision: 7, AuthorizationIssuerEpoch: 3, AuthorizationRevision: 5, Phase: string(model.AckVerified), ValidUntil: &until}, + {MemberID: "cluster", WriterEpoch: 2, Revision: 7, AuthorizationIssuerEpoch: 3, AuthorizationRevision: 4, Phase: string(model.AckVerified), ValidUntil: &until}, + } + if s.verified(rows, now, "resolver", 7, 2, 3, 5) { + t.Fatal("old cluster authorization acknowledgement satisfied a renewed lease") + } + rows[1].AuthorizationRevision = 5 + if !s.verified(rows, now, "resolver", 7, 2, 3, 5) { + t.Fatal("fresh complete member set was not ready") + } + rows[1].WriterEpoch = 1 + if s.verified(rows, now, "resolver", 7, 2, 3, 5) { + t.Fatal("retired planner epoch satisfied current readiness") + } +} + +func TestReadinessExpiresWithoutAnotherAck(t *testing.T) { + now := time.Now().UTC() + until := metav1.NewTime(now.Add(-time.Second)) + b := &dnsv1alpha1.DNSResolverBinding{ObjectMeta: metav1.ObjectMeta{Name: "binding", Namespace: "platform", Labels: map[string]string{"internal-dns.miloapis.com/region": "east", "internal-dns.miloapis.com/shard": "shared"}}, Spec: dnsv1alpha1.DNSResolverBindingSpec{Configuration: dnsv1alpha1.DNSResolverBindingConfiguration{Revision: 1}, Authorization: dnsv1alpha1.DNSResolverAccessAuthorization{WriterEpoch: 1, Sequence: 1, ValidUntil: metav1.NewTime(now.Add(time.Minute))}}, Status: dnsv1alpha1.DNSResolverBindingStatus{Phase: "Serving", MemberAcknowledgements: []dnsv1alpha1.DNSApplyAcknowledgement{{MemberID: "resolver", WriterEpoch: 1, Revision: 1, AuthorizationIssuerEpoch: 1, AuthorizationRevision: 1, Phase: string(model.AckVerified), ValidUntil: &until}}}} + cl := testClient(t, b) + s := AckSink{Client: cl, Namespace: "platform", Region: "east", Shard: "shared", Members: []model.Member{{MemberID: "resolver", Role: "resolver"}}, Now: func() time.Time { return now }} + if err := s.Refresh(context.Background()); err != nil { + t.Fatal(err) + } + var actual dnsv1alpha1.DNSResolverBinding + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(b), &actual); err != nil { + t.Fatal(err) + } + if actual.Status.Phase == "Serving" { + t.Fatal("dead member retained ready status beyond its verification lease") + } +} + +func TestServingMinimumPublicationFenceSurvivesSupersededActivation(t *testing.T) { + ns := "platform" + old := &dnsv1alpha1.DNSPublicationManifest{ObjectMeta: metav1.ObjectMeta{Name: "old-manifest", Namespace: ns}, Spec: dnsv1alpha1.DNSPublicationManifestSpec{ZoneRef: dnsv1alpha1.DNSObjectReference{UID: "zone"}, WriterEpoch: 1, Revision: 1}} + current := old.DeepCopy() + current.Name = "current-manifest" + current.Spec.Revision = 2 + own := &dnsv1alpha1.DNSPublicationOwnership{ObjectMeta: metav1.ObjectMeta{Name: "owner-" + model.OpaqueToken("zone")[:20], Namespace: ns}, Spec: dnsv1alpha1.DNSPublicationOwnershipSpec{ZoneUID: "zone", WriterEpoch: 1, ActiveManifestName: current.Name}} + dep := &dnsv1alpha1.DNSTransportOutbox{ObjectMeta: metav1.ObjectMeta{Name: "old-activation", Namespace: ns}, Spec: dnsv1alpha1.DNSTransportOutboxSpec{ManifestRef: dnsv1alpha1.DNSObjectReference{Name: old.Name}, WriterEpoch: 1, Revision: 1}} + newer := &dnsv1alpha1.DNSTransportOutbox{ObjectMeta: metav1.ObjectMeta{Name: model.PublicationActivationName(current.Name, "east", "shared"), Namespace: ns}, Status: dnsv1alpha1.DNSTransportOutboxStatus{State: "Acknowledged"}} + o := Outbox{Client: testClient(t, old, current, own, dep, newer), Namespace: ns} + env := model.Envelope{Kind: model.KindServingSnapshot, Region: "east", Shard: "shared"} + ready, err := o.dependencyReady(context.Background(), env, dep.Name, map[string]string{}) + if err != nil || !ready { + t.Fatalf("newer committed publication did not satisfy the minimum fence: %v", err) + } + env.Kind = model.KindPublicationManifest + ready, err = o.dependencyReady(context.Background(), env, dep.Name, map[string]string{}) + if err != nil || ready { + t.Fatal("manifest allowed mixing chunk dependencies from other revisions") + } +} + +func TestPublicationReadinessRequiresEveryRegionalMember(t *testing.T) { + now := time.Now().UTC() + until := metav1.NewTime(now.Add(time.Minute)) + s := AckSink{PublicationRegions: map[string][]model.Member{"east": {{MemberID: "east-a", Role: "regional"}, {MemberID: "east-b", Role: "regional"}}, "west": {{MemberID: "west-a", Role: "regional"}, {MemberID: "west-b", Role: "regional"}}}} + row := func(member string) dnsv1alpha1.DNSApplyAcknowledgement { + return dnsv1alpha1.DNSApplyAcknowledgement{MemberID: member, WriterEpoch: 1, Revision: 2, Phase: string(model.AckVerified), ValidUntil: &until} + } + rows := make([]dnsv1alpha1.DNSApplyAcknowledgement, 0, 4) + rows = append(rows, row("east-a"), row("east-b")) + if s.verified(rows, now, "publication", 2, 1, 0, 0) { + t.Fatal("east replicas satisfied missing west region") + } + rows = append(rows, row("west-a")) + if s.verified(rows, now, "publication", 2, 1, 0, 0) { + t.Fatal("missing regional member acknowledged publication") + } + rows = append(rows, row("west-b")) + if !s.verified(rows, now, "publication", 2, 1, 0, 0) { + t.Fatal("all regional members did not satisfy publication") + } +} + +func TestOutOfOrderAckDoesNotReplaceWithdrawalInSameSecond(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + expired := dnsv1alpha1.DNSApplyAcknowledgement{MemberID: "auth", Phase: string(model.AckExpired), ObservedAt: metav1.NewTime(now), ObservedUnixNano: now.Add(900 * time.Millisecond).UnixNano()} + late := expired + late.Phase = string(model.AckVerified) + late.ObservedUnixNano = now.Add(100 * time.Millisecond).UnixNano() + rows := upsertAck([]dnsv1alpha1.DNSApplyAcknowledgement{expired}, late) + if rows[0].Phase != string(model.AckExpired) { + t.Fatal("reordered positive acknowledgement overwrote a newer withdrawal") + } +} + +func TestBindingReadinessRequiresCurrentCommittedSnapshot(t *testing.T) { + now := time.Now().UTC() + until := metav1.NewTime(now.Add(time.Minute)) + b := &dnsv1alpha1.DNSResolverBinding{ObjectMeta: metav1.ObjectMeta{UID: "binding-uid"}, Spec: dnsv1alpha1.DNSResolverBindingSpec{Configuration: dnsv1alpha1.DNSResolverBindingConfiguration{Generation: 1, Revision: 7}, Authorization: dnsv1alpha1.DNSResolverAccessAuthorization{WriterEpoch: 3, Sequence: 5, ValidUntil: until}}, Status: dnsv1alpha1.DNSResolverBindingStatus{MemberAcknowledgements: []dnsv1alpha1.DNSApplyAcknowledgement{ + {MemberID: "node", WriterEpoch: 2, SnapshotRevision: 8, Revision: 7, AuthorizationIssuerEpoch: 3, AuthorizationRevision: 5, Phase: string(model.AckVerified), ValidUntil: &until}, + {MemberID: "cluster", WriterEpoch: 2, SnapshotRevision: 8, Revision: 7, AuthorizationIssuerEpoch: 3, AuthorizationRevision: 5, Phase: string(model.AckVerified), ValidUntil: &until}, + }}} + s := AckSink{Members: []model.Member{{MemberID: "node", Role: "node"}, {MemberID: "cluster", Role: "cluster"}}} + envelope := &model.Envelope{Epoch: 2, Revision: 9} + snapshot := &model.ServingSnapshot{Bindings: []model.Binding{{BindingUID: "binding-uid", BindingGeneration: 1, ConfigurationRevision: 7, Authorization: model.Authorization{IssuerEpoch: 3, Revision: 5}}}} + if s.bindingVerified(b, envelope, snapshot, now) { + t.Fatal("retired snapshot remained ready") + } + b.Status.MemberAcknowledgements[0].SnapshotRevision = 9 + if s.bindingVerified(b, envelope, snapshot, now) { + t.Fatal("mixed snapshot acknowledgements remained ready") + } + b.Status.MemberAcknowledgements[1].SnapshotRevision = 9 + if !s.bindingVerified(b, envelope, snapshot, now) { + t.Fatal("current complete member set was not ready") + } + snapshot.Bindings = nil + if s.bindingVerified(b, envelope, snapshot, now) { + t.Fatal("omitted binding remained ready") + } +} + +func TestExpiredAcknowledgementReplayIsConsumedWithoutStatus(t *testing.T) { + now := time.Now().UTC() + s := AckSink{Namespace: "platform", Region: "east", Shard: "shared", Members: []model.Member{{MemberID: "auth", ReplicaID: "replica", Role: "regional"}}, Now: func() time.Time { return now }} + ack := model.MemberAck{MemberID: "auth", ReplicaID: "replica", ResourceUID: "zone", Kind: model.KindPublicationManifest, Epoch: 1, Revision: 1, Phase: model.AckVerified, ObservedAt: now.Add(-time.Minute), ValidUntil: now.Add(-time.Second)} + env, err := model.NewEnvelope(model.KindMemberAck, "old-health", "east", "shared", "zone", 1, 1, now.Add(-time.Minute), ack) + if err != nil { + t.Fatal(err) + } + // No API client is provided: expired replay must be consumed without any + // attempted status write, even when project APIs are unavailable. + if err := s.Apply(context.Background(), model.AckSubject("east", "shared", "auth"), env); err != nil { + t.Fatalf("expired replay became permanently retryable: %v", err) + } +} + +func zoneReferences(uids []types.UID) []dnsv1alpha1.DNSObjectReference { + refs := make([]dnsv1alpha1.DNSObjectReference, 0, len(uids)) + for _, uid := range uids { + refs = append(refs, dnsv1alpha1.DNSObjectReference{Name: string(uid), UID: uid}) + } + return refs +} + +func TestZeroTargetTombstoneNeedsNoReplicaACK(t *testing.T) { + sink := &AckSink{} + manifest := &dnsv1alpha1.DNSPublicationManifest{} + if sink.publicationVerified(manifest, time.Now()) { + t.Fatal("active zero-target publication must not be declared served") + } + manifest.Spec.Tombstone = true + if !sink.publicationVerified(manifest, time.Now()) { + t.Fatal("zero-target tombstone must complete") + } + manifest.Spec.ServingTargets = []dnsv1alpha1.DNSPublicationTarget{{Region: "retired-region", Shard: "retired-shard"}} + if sink.publicationVerified(manifest, time.Now()) { + t.Fatal("prior-target withdrawal must await its ACKs") + } +} diff --git a/internal/internaldns/platform/values.go b/internal/internaldns/platform/values.go new file mode 100644 index 00000000..9af3762e --- /dev/null +++ b/internal/internaldns/platform/values.go @@ -0,0 +1,12 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package platform + +const dnsValuePending = "Pending" +const dnsValueAcknowledged = "Acknowledged" +const dnsValueResolver = "resolver" +const dnsValueCluster = "cluster" +const dnsValueRegional = "regional" +const dnsValueSuperseded = "Superseded" + +const dnsValueServing = "Serving" From 5b170651b2e0c398c876e56ea8a8e84feea2a2cc Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Wed, 21 Oct 2026 20:12:08 -0500 Subject: [PATCH 2/3] fix: preserve serving state during delivery limits --- internal/internaldns/platform/capacity.go | 43 ++++--- .../internaldns/platform/capacity_test.go | 10 +- .../platform/delivery_review_test.go | 115 ++++++++++++++++++ internal/internaldns/platform/planner.go | 80 +++++++++--- 4 files changed, 215 insertions(+), 33 deletions(-) create mode 100644 internal/internaldns/platform/delivery_review_test.go diff --git a/internal/internaldns/platform/capacity.go b/internal/internaldns/platform/capacity.go index d729cd1c..f7992591 100644 --- a/internal/internaldns/platform/capacity.go +++ b/internal/internaldns/platform/capacity.go @@ -5,6 +5,8 @@ package platform import ( "context" "encoding/json" + "errors" + "fmt" "sort" "time" @@ -13,7 +15,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) -func fitSnapshot(snapshot model.ServingSnapshot, prior model.ServingSnapshot, maxBindings, maxBytes int) (model.ServingSnapshot, map[string]bool) { +func fitSnapshot(snapshot model.ServingSnapshot, prior model.ServingSnapshot, maxBindings, maxBytes int) (model.ServingSnapshot, map[string]bool, error) { previous := map[string]model.Binding{} for _, binding := range prior.Bindings { previous[binding.BindingUID] = binding @@ -27,20 +29,34 @@ func fitSnapshot(snapshot model.ServingSnapshot, prior model.ServingSnapshot, ma } return candidates[i].BindingUID < candidates[j].BindingUID }) - snapshot.Bindings = nil + snapshot.Bindings = []model.Binding{} rejected := map[string]bool{} + envelope, err := model.NewEnvelope(model.KindServingSnapshot, "dns-serving-"+model.OpaqueToken(snapshot.Region+"/"+snapshot.Shard)+"-e18446744073709551615-r18446744073709551615", snapshot.Region, snapshot.Shard, "00000000-0000-0000-0000-000000000000", snapshot.ConfigurationEpoch, snapshot.ConfigurationRevision, time.Now(), snapshot) + if err != nil { + return snapshot, rejected, err + } + payload, err := json.Marshal(envelope) + if err != nil { + return snapshot, rejected, err + } + used := len(payload) fits := func(binding model.Binding) bool { if len(snapshot.Bindings) >= maxBindings { return false } - candidate := snapshot - candidate.Bindings = append(append([]model.Binding(nil), snapshot.Bindings...), binding) - envelope, err := model.NewEnvelope(model.KindServingSnapshot, "dns-serving-"+model.OpaqueToken(candidate.Region+"/"+candidate.Shard)+"-e18446744073709551615-r18446744073709551615", candidate.Region, candidate.Shard, "00000000-0000-0000-0000-000000000000", candidate.ConfigurationEpoch, candidate.ConfigurationRevision, time.Now(), candidate) + data, err := json.Marshal(binding) if err != nil { return false } - payload, err := json.Marshal(envelope) - return err == nil && len(payload) <= maxBytes + size := len(data) + if len(snapshot.Bindings) > 0 { + size++ + } + if used+size > maxBytes { + return false + } + used += size + return true } for _, binding := range candidates { if fits(binding) { @@ -53,15 +69,11 @@ func fitSnapshot(snapshot model.ServingSnapshot, prior model.ServingSnapshot, ma snapshot.Bindings = append(snapshot.Bindings, old) continue } - binding.ConfigurationPending = true - binding.Zones = nil - if fits(binding) { - snapshot.Bindings = append(snapshot.Bindings, binding) - } + return prior, rejected, fmt.Errorf("capacity cannot preserve established binding %s", binding.BindingUID) } } sort.Slice(snapshot.Bindings, func(i, j int) bool { return snapshot.Bindings[i].BindingUID < snapshot.Bindings[j].BindingUID }) - return snapshot, rejected + return snapshot, rejected, nil } func safeExpansion(old, candidate model.Binding) bool { @@ -83,6 +95,7 @@ func safeExpansion(old, candidate model.Binding) bool { } func (p *Planner) reportCapacity(ctx context.Context, bindings []dnsv1alpha1.DNSResolverBinding, rejected map[string]bool, now time.Time) error { + var failures []error for i := range bindings { binding := &bindings[i] if binding.Spec.Placement.Region != p.Config.Region || binding.Spec.Placement.Shard != p.Config.Shard { @@ -92,9 +105,9 @@ func (p *Planner) reportCapacity(ctx context.Context, bindings []dnsv1alpha1.DNS condition(&binding.Status.Conditions, "CapacityAvailable", !rejected[string(binding.UID)], "ShardCapacity", binding.Generation, now) if !jsonEqualStatus(base.Status, binding.Status) { if err := p.Client.Status().Patch(ctx, binding, client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})); err != nil { - return err + failures = append(failures, err) } } } - return nil + return errors.Join(failures...) } diff --git a/internal/internaldns/platform/capacity_test.go b/internal/internaldns/platform/capacity_test.go index 7cf95fab..c3ce73a1 100644 --- a/internal/internaldns/platform/capacity_test.go +++ b/internal/internaldns/platform/capacity_test.go @@ -23,7 +23,10 @@ func TestCapacityOverflowDoesNotBlockWithdrawal(t *testing.T) { candidate := prior candidate.ConfigurationRevision++ candidate.Bindings = []model.Binding{capacityBinding(2), capacityBinding(3), capacityBinding(4)} - bounded, rejected := fitSnapshot(candidate, prior, 2, 512*1024) + bounded, rejected, err := fitSnapshot(candidate, prior, 2, 512*1024) + if err != nil { + t.Fatal(err) + } if len(bounded.Bindings) != 2 || len(rejected) != 1 { t.Fatalf("capacity result: %#v, rejected %#v", bounded.Bindings, rejected) } @@ -46,7 +49,10 @@ func TestByteOverflowPreservesExistingContext(t *testing.T) { } binding.ConfigurationRevision++ candidate.Bindings = []model.Binding{binding, capacityBinding(2)} - bounded, rejected := fitSnapshot(candidate, prior, 2, 4096) + bounded, rejected, err := fitSnapshot(candidate, prior, 2, 4096) + if err != nil { + t.Fatal(err) + } if !rejected["binding-1"] || len(bounded.Bindings) != 2 || bounded.Bindings[0].ConfigurationRevision != 1 { t.Fatalf("overflow interrupted an established context: %#v", bounded) } diff --git a/internal/internaldns/platform/delivery_review_test.go b/internal/internaldns/platform/delivery_review_test.go new file mode 100644 index 00000000..2baba374 --- /dev/null +++ b/internal/internaldns/platform/delivery_review_test.go @@ -0,0 +1,115 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package platform + +import ( + "context" + "encoding/json" + "fmt" + "testing" + "time" + + dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" + "go.miloapis.com/dns-operator/internal/internaldns/model" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +func TestCapacityReductionCannotDropEstablishedBindings(t *testing.T) { + prior := model.ServingSnapshot{Region: "east", Shard: "a", ConfigurationEpoch: 1, ConfigurationRevision: 1, GeneratedAt: time.Now(), Bindings: []model.Binding{capacityBinding(1), capacityBinding(2)}} + for _, limits := range [][2]int{{1, 512 * 1024}, {2, 1}} { + _, _, err := fitSnapshot(prior, prior, limits[0], limits[1]) + if err == nil { + t.Fatalf("accepted limits %v that drop an established context", limits) + } + } +} + +func TestPlannerWaitsForPublicationActivation(t *testing.T) { + ctx := context.Background() + now := time.Now().UTC() + ns, region, shard := "platform", "east", "a" + state := shardState{Holder: "planner", Epoch: 1, NextRevision: 1, LeaseUntil: now.Add(time.Minute)} + data, _ := json.Marshal(state) + owner := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "dns-shard-owner-" + model.OpaqueToken(region+"/"+shard), Namespace: ns, UID: "owner"}, Data: map[string]string{"state": string(data)}} + zone := &dnsv1alpha1.DNSPublicationOwnership{ObjectMeta: metav1.ObjectMeta{Name: "owner-" + model.OpaqueToken("zone")[:20], Namespace: ns}, Spec: dnsv1alpha1.DNSPublicationOwnershipSpec{ZoneUID: "zone", WriterEpoch: 1, ActiveManifestName: "manifest"}} + manifest := &dnsv1alpha1.DNSPublicationManifest{ObjectMeta: metav1.ObjectMeta{Name: "manifest", Namespace: ns}, Spec: dnsv1alpha1.DNSPublicationManifestSpec{ZoneRef: dnsv1alpha1.DNSObjectReference{UID: "zone"}, ZoneApex: "datum.internal", WriterEpoch: 1, Revision: 2, ServingTargets: []dnsv1alpha1.DNSPublicationTarget{{Region: region, Shard: shard}}}} + binding := &dnsv1alpha1.DNSResolverBinding{ObjectMeta: metav1.ObjectMeta{Name: "binding", Namespace: ns, UID: "binding"}, Spec: dnsv1alpha1.DNSResolverBindingSpec{Source: dnsv1alpha1.DNSResolverBindingSource{ProjectUID: "project", ResolverContextRef: dnsv1alpha1.DNSObjectReference{UID: "context"}}, Placement: dnsv1alpha1.DNSResolverBindingPlacement{Region: region, Shard: shard}, Version: dnsv1alpha1.DNSResolverAccessVersion{Epoch: 1, Sequence: 1}, Configuration: dnsv1alpha1.DNSResolverBindingConfiguration{Generation: 1, Revision: 1, Query: dnsv1alpha1.DNSResolverBindingQuery{NodeAddress: "fd53::1", RegionalAddress: "fd54::1"}, ZoneRefs: []dnsv1alpha1.DNSObjectReference{{UID: "zone"}}}}} + binding.Labels = map[string]string{"internal-dns.miloapis.com/region": model.SafeToken(region), "internal-dns.miloapis.com/shard": model.SafeToken(shard)} + cl := testClient(t, owner, zone, manifest, binding) + p := Planner{Client: capacityConflictClient{Client: cl}, Allocator: &Allocator{Client: cl, Namespace: ns}, Config: PlannerConfig{Namespace: ns, Region: region, Shard: shard, Identity: "planner"}, Now: func() time.Time { return now }} + if err := p.Step(ctx); err != nil { + t.Fatal(err) + } + var outs dnsv1alpha1.DNSTransportOutboxList + if err := cl.List(ctx, &outs); err != nil { + t.Fatal(err) + } + if len(outs.Items) != 1 { + t.Fatalf("outboxes=%d", len(outs.Items)) + } + out := outs.Items[0] + activation := model.PublicationActivationName(manifest.Name, region, shard) + if len(out.Spec.DependsOn) != 1 || out.Spec.DependsOn[0] != activation { + t.Fatalf("activation fence missing: %v", out.Spec.DependsOn) + } + var envelope model.Envelope + var snapshot model.ServingSnapshot + _ = json.Unmarshal(out.Spec.Payload, &envelope) + _ = json.Unmarshal(envelope.Payload, &snapshot) + if len(snapshot.RequiredPublications) != 1 || snapshot.RequiredPublications[0].Revision != 2 { + t.Fatalf("publication fence missing: %+v", snapshot.RequiredPublications) + } + worker := Outbox{Client: cl, Namespace: ns} + ready, err := worker.dependencyReady(ctx, envelope, activation, map[string]string{}) + if err != nil || ready { + t.Fatalf("undelivered activation accepted: %v %v", ready, err) + } + dependency := &dnsv1alpha1.DNSTransportOutbox{ObjectMeta: metav1.ObjectMeta{Name: activation, Namespace: ns}, Spec: dnsv1alpha1.DNSTransportOutboxSpec{ManifestRef: dnsv1alpha1.DNSObjectReference{Name: manifest.Name}}} + if err := cl.Create(ctx, dependency); err != nil { + t.Fatal(err) + } + ready, err = worker.dependencyReady(ctx, envelope, activation, map[string]string{}) + if err != nil || ready { + t.Fatalf("pending activation accepted: %v %v", ready, err) + } + dependency.Status.State = dnsValueAcknowledged + if err := cl.Status().Update(ctx, dependency); err != nil { + t.Fatal(err) + } + ready, err = worker.dependencyReady(ctx, envelope, activation, map[string]string{}) + if err != nil || !ready { + t.Fatalf("delivered activation rejected: %v %v", ready, err) + } + var before corev1.ConfigMap + if err := cl.Get(ctx, client.ObjectKeyFromObject(owner), &before); err != nil { + t.Fatal(err) + } + p.Config.MaxSnapshotBytes = 1 + if err := p.Step(ctx); err == nil { + t.Fatal("oversized envelope accepted") + } + var after corev1.ConfigMap + if err := cl.Get(ctx, client.ObjectKeyFromObject(owner), &after); err != nil { + t.Fatal(err) + } + if before.Data["state"] != after.Data["state"] { + t.Fatal("size rejection consumed a revision") + } + +} + +type capacityConflictClient struct{ client.Client } +type capacityConflictStatus struct{ client.SubResourceWriter } + +func (c capacityConflictClient) Status() client.SubResourceWriter { + return capacityConflictStatus{SubResourceWriter: c.Client.Status()} +} + +func (c capacityConflictStatus) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if _, ok := obj.(*dnsv1alpha1.DNSResolverBinding); ok { + return fmt.Errorf("injected binding status conflict") + } + return c.SubResourceWriter.Patch(ctx, obj, patch, opts...) +} diff --git a/internal/internaldns/platform/planner.go b/internal/internaldns/platform/planner.go index ef943428..31396da1 100644 --- a/internal/internaldns/platform/planner.go +++ b/internal/internaldns/platform/planner.go @@ -16,6 +16,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" ) const shardOwnerLabel = "internal-dns.miloapis.com/shard-owner" @@ -132,6 +133,7 @@ func (p *Planner) Step(ctx context.Context) error { return err } } + activations := map[string]string{} priorBindings := map[string]model.Binding{} for _, b := range prior.Bindings { priorBindings[b.BindingUID] = b @@ -187,6 +189,7 @@ func (p *Planner) Step(ctx context.Context) error { } } } + activations[string(uid)] = model.PublicationActivationName(m.Name, c.Region, c.Shard) view.Zones = append(view.Zones, attachment) } if !complete { @@ -210,9 +213,45 @@ func (p *Planner) Step(ctx context.Context) error { maxBytes = 512 * 1024 } var rejected map[string]bool - snapshot, rejected = fitSnapshot(snapshot, prior, maxBindings, maxBytes) - if err := p.reportCapacity(ctx, bindings.Items, rejected, now); err != nil { - return err + var fitErr error + snapshot, rejected, fitErr = fitSnapshot(snapshot, prior, maxBindings, maxBytes) + if fitErr != nil { + return fitErr + } + dependencies := map[string]bool{} + requirements := map[string]model.PublicationRequirement{} + for _, binding := range snapshot.Bindings { + for _, zone := range binding.Zones { + if zone.Pending { + continue + } + activation := activations[zone.ZoneUID] + if activation == "" { + return fmt.Errorf("zone %s has no publication activation", zone.ZoneUID) + } + dependencies[activation] = true + requirement := model.PublicationRequirement{ZoneUID: zone.ZoneUID, Epoch: zone.RequiredPublicationEpoch, Revision: zone.RequiredPublicationRevision} + old := requirements[zone.ZoneUID] + if model.Compare(requirement.Epoch, requirement.Revision, old.Epoch, old.Revision) > 0 { + requirements[zone.ZoneUID] = requirement + } + } + } + var dependsOn []string + for dependency := range dependencies { + dependsOn = append(dependsOn, dependency) + } + sort.Strings(dependsOn) + for _, requirement := range requirements { + snapshot.RequiredPublications = append(snapshot.RequiredPublications, requirement) + } + sort.Slice(snapshot.RequiredPublications, func(i, j int) bool { + return snapshot.RequiredPublications[i].ZoneUID < snapshot.RequiredPublications[j].ZoneUID + }) + report := func() { + if err := p.reportCapacity(ctx, bindings.Items, rejected, now); err != nil { + log.FromContext(ctx).Error(err, "report shard capacity") + } } if err := snapshot.Validate(); err != nil { return err @@ -225,27 +264,22 @@ func (p *Planner) Step(ctx context.Context) error { hash := model.Hash(b) if state.ContentHash == hash { if state.LeaseUntil.Sub(now) > lease/2 { + report() return nil } state.LeaseUntil = now.Add(lease) b, _ := json.Marshal(state) owner.Data["state"] = string(b) - return p.Client.Update(ctx, &owner) + if err := p.Client.Update(ctx, &owner); err != nil { + return err + } + report() + return nil } if !state.LeaseUntil.After(now) { return fmt.Errorf("shared planner lost its lease while staging snapshot") } - // Reserve the revision before writing immutable artifacts. A crash after - // staging leaves an orphan, and retry uses a fresh revision instead of trying - // to rewrite an immutable event with a different timestamp or desired state. revision := state.NextRevision - state.NextRevision++ - state.LeaseUntil = now.Add(lease) - reserved, _ := json.Marshal(state) - owner.Data["state"] = string(reserved) - if err := p.Client.Update(ctx, &owner); err != nil { - return err - } name := fmt.Sprintf("dns-serving-%s-e%d-r%d", model.OpaqueToken(c.Region+"/"+c.Shard), state.Epoch, revision) env, err := model.NewEnvelope(model.KindServingSnapshot, name, c.Region, c.Shard, string(owner.UID), state.Epoch, revision, now, snapshot) if err != nil { @@ -255,7 +289,17 @@ func (p *Planner) Step(ctx context.Context) error { if len(payload) > maxBytes { return fmt.Errorf("shard serving envelope exceeds configured transport budget: %d > %d", len(payload), maxBytes) } - out := &dnsv1alpha1.DNSTransportOutbox{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: c.Namespace, Labels: map[string]string{shardOwnerLabel: owner.Name, "internal-dns.miloapis.com/region": model.SafeToken(c.Region), "internal-dns.miloapis.com/shard": model.SafeToken(c.Shard)}}, Spec: dnsv1alpha1.DNSTransportOutboxSpec{Subject: model.ServingSubject(c.Region, c.Shard), ResourceUID: owner.UID, WriterEpoch: int64(state.Epoch), Revision: int64(revision), PayloadHash: model.Hash(payload), Payload: payload, Activation: true, DependsOn: nil}} + // Reserve the revision before writing immutable artifacts. A crash after + // staging leaves an orphan, and retry uses a fresh revision instead of trying + // to rewrite an immutable event with a different timestamp or desired state. + state.NextRevision++ + state.LeaseUntil = now.Add(lease) + reserved, _ := json.Marshal(state) + owner.Data["state"] = string(reserved) + if err := p.Client.Update(ctx, &owner); err != nil { + return err + } + out := &dnsv1alpha1.DNSTransportOutbox{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: c.Namespace, Labels: map[string]string{shardOwnerLabel: owner.Name, "internal-dns.miloapis.com/region": model.SafeToken(c.Region), "internal-dns.miloapis.com/shard": model.SafeToken(c.Shard)}}, Spec: dnsv1alpha1.DNSTransportOutboxSpec{Subject: model.ServingSubject(c.Region, c.Shard), ResourceUID: owner.UID, WriterEpoch: int64(state.Epoch), Revision: int64(revision), PayloadHash: model.Hash(payload), Payload: payload, Activation: true, DependsOn: dependsOn}} if err := p.Client.Create(ctx, out); apierrors.IsAlreadyExists(err) { var existing dnsv1alpha1.DNSTransportOutbox if err := p.Client.Get(ctx, client.ObjectKeyFromObject(out), &existing); err != nil { @@ -273,5 +317,9 @@ func (p *Planner) Step(ctx context.Context) error { b, _ = json.Marshal(state) owner.Data["state"] = string(b) // Publication is enabled only after this resourceVersion-checked commit. - return p.Client.Update(ctx, &owner) + if err := p.Client.Update(ctx, &owner); err != nil { + return err + } + report() + return nil } From 42d875f00fbcef38414d21573830feb96886aa68 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Wed, 21 Oct 2026 20:20:16 -0500 Subject: [PATCH 3/3] refactor: separate publication dependency planning --- internal/internaldns/platform/planner.go | 66 +++++++++++++----------- 1 file changed, 37 insertions(+), 29 deletions(-) diff --git a/internal/internaldns/platform/planner.go b/internal/internaldns/platform/planner.go index 31396da1..08023946 100644 --- a/internal/internaldns/platform/planner.go +++ b/internal/internaldns/platform/planner.go @@ -218,36 +218,10 @@ func (p *Planner) Step(ctx context.Context) error { if fitErr != nil { return fitErr } - dependencies := map[string]bool{} - requirements := map[string]model.PublicationRequirement{} - for _, binding := range snapshot.Bindings { - for _, zone := range binding.Zones { - if zone.Pending { - continue - } - activation := activations[zone.ZoneUID] - if activation == "" { - return fmt.Errorf("zone %s has no publication activation", zone.ZoneUID) - } - dependencies[activation] = true - requirement := model.PublicationRequirement{ZoneUID: zone.ZoneUID, Epoch: zone.RequiredPublicationEpoch, Revision: zone.RequiredPublicationRevision} - old := requirements[zone.ZoneUID] - if model.Compare(requirement.Epoch, requirement.Revision, old.Epoch, old.Revision) > 0 { - requirements[zone.ZoneUID] = requirement - } - } - } - var dependsOn []string - for dependency := range dependencies { - dependsOn = append(dependsOn, dependency) - } - sort.Strings(dependsOn) - for _, requirement := range requirements { - snapshot.RequiredPublications = append(snapshot.RequiredPublications, requirement) + dependsOn, err := publicationDependencies(&snapshot, activations) + if err != nil { + return err } - sort.Slice(snapshot.RequiredPublications, func(i, j int) bool { - return snapshot.RequiredPublications[i].ZoneUID < snapshot.RequiredPublications[j].ZoneUID - }) report := func() { if err := p.reportCapacity(ctx, bindings.Items, rejected, now); err != nil { log.FromContext(ctx).Error(err, "report shard capacity") @@ -323,3 +297,37 @@ func (p *Planner) Step(ctx context.Context) error { report() return nil } + +func publicationDependencies(snapshot *model.ServingSnapshot, activations map[string]string) ([]string, error) { + dependencies := map[string]bool{} + requirements := map[string]model.PublicationRequirement{} + for _, binding := range snapshot.Bindings { + for _, zone := range binding.Zones { + if zone.Pending { + continue + } + activation := activations[zone.ZoneUID] + if activation == "" { + return nil, fmt.Errorf("zone %s has no publication activation", zone.ZoneUID) + } + dependencies[activation] = true + requirement := model.PublicationRequirement{ZoneUID: zone.ZoneUID, Epoch: zone.RequiredPublicationEpoch, Revision: zone.RequiredPublicationRevision} + old := requirements[zone.ZoneUID] + if model.Compare(requirement.Epoch, requirement.Revision, old.Epoch, old.Revision) > 0 { + requirements[zone.ZoneUID] = requirement + } + } + } + var dependsOn []string + for dependency := range dependencies { + dependsOn = append(dependsOn, dependency) + } + sort.Strings(dependsOn) + for _, requirement := range requirements { + snapshot.RequiredPublications = append(snapshot.RequiredPublications, requirement) + } + sort.Slice(snapshot.RequiredPublications, func(i, j int) bool { + return snapshot.RequiredPublications[i].ZoneUID < snapshot.RequiredPublications[j].ZoneUID + }) + return dependsOn, nil +}