From 9a4046e3e25a56da72a4477b0c0172eaa9300168 Mon Sep 17 00:00:00 2001 From: "beug.k@beugsiud.de" Date: Tue, 13 Oct 2026 20:07:35 +0200 Subject: [PATCH 1/2] feat(agent): push deployments by polling, apply each deployment once - GetDesiredStateResponse.deployment_id set: the desired revision is applied as a push with that ID, also in manual mode (ADR 0022). - The reports of the last 20 push deployments are kept in state.json. A resent push with a handled deployment ID is not applied again; its report is delivered again unless it is still queued. - A push already queued with the same deployment ID is not queued twice. - Reports the server refuses for good (InvalidArgument, NotFound, AlreadyExists, FailedPrecondition, OutOfRange) are dropped instead of blocking the report queue. - Tests for polling pushes, resent pushes, mode changes via a new Welcome, persistence and bounds of handled deployments. Co-Authored-By: Claude Opus 5.5 --- agent/core/agent.go | 62 +++++- agent/core/agent_test.go | 3 +- agent/core/helpers_test.go | 23 ++- agent/core/push_test.go | 366 ++++++++++++++++++++++++++++++++++++ agent/core/server_source.go | 27 ++- agent/core/source.go | 14 +- agent/core/store.go | 46 ++++- 7 files changed, 528 insertions(+), 13 deletions(-) create mode 100644 agent/core/push_test.go diff --git a/agent/core/agent.go b/agent/core/agent.go index 58a5ed0..13f5a10 100644 --- a/agent/core/agent.go +++ b/agent/core/agent.go @@ -121,12 +121,14 @@ func (a *Agent) Run(ctx context.Context) error { return err } -// state returns a copy of the state (without the report queue). +// state returns a copy of the state (without the report queue and the +// handled deployments). func (a *Agent) state() State { a.mu.Lock() defer a.mu.Unlock() st := *a.st st.PendingReports = nil + st.HandledDeployments = nil return st } @@ -195,7 +197,8 @@ func (a *Agent) refreshStatus(ctx context.Context) { } // enqueue adds a request. A pull is dropped if an equal or newer revision -// is already queued; a push replaces queued pulls up to its revision. +// is already queued, a push if the same deployment is already queued; a +// push replaces queued pulls up to its revision. func (a *Agent) enqueue(r applyRequest) { a.qmu.Lock() if r.trigger == TriggerPull { @@ -206,6 +209,12 @@ func (a *Agent) enqueue(r applyRequest) { } } } else { + if r.deploymentID != "" && slices.ContainsFunc(a.queue, func(q applyRequest) bool { + return q.trigger == TriggerPush && q.deploymentID == r.deploymentID + }) { + a.qmu.Unlock() + return + } a.queue = slices.DeleteFunc(a.queue, func(q applyRequest) bool { return q.trigger == TriggerPull && q.revision <= r.revision }) @@ -247,8 +256,11 @@ func (a *Agent) applyLoop(ctx context.Context) { // handle decides whether a request leads to an apply. func (a *Agent) handle(ctx context.Context, r applyRequest) { - st := a.state() log := a.log.With("revision", r.revision, "trigger", r.trigger, "deployment_id", r.deploymentID) + if r.trigger == TriggerPush && a.skipHandled(log, r) { + return + } + st := a.state() switch { case r.trigger == TriggerPull && r.revision <= st.Applied: log.Debug("revision already applied or older", "applied", st.Applied) @@ -270,6 +282,38 @@ func (a *Agent) handle(ctx context.Context, r applyRequest) { a.apply(ctx, r, st, log) } +// skipHandled reports whether a push belongs to a deployment that was +// handled before, which the server resends after a reconnect or announces +// while polling as long as it has no report. The revision is not applied +// again: if the report is still queued, the report loop delivers it; +// otherwise the remembered report is queued again so that the server can +// close the deployment. +func (a *Agent) skipHandled(log *slog.Logger, r applyRequest) bool { + a.mu.Lock() + prev, ok := a.st.handledDeployment(r.deploymentID) + if !ok { + a.mu.Unlock() + return false + } + queued := slices.ContainsFunc(a.st.PendingReports, func(p DeploymentReport) bool { + return p.DeploymentID == r.deploymentID + }) + var err error + if !queued { + a.st.PendingReports = append(a.st.PendingReports, prev) + a.st.UpdatedAt = time.Now().UTC() + err = a.store.SaveState(a.st) + } + a.mu.Unlock() + if err != nil { + log.Error("cannot save state", "error", err) + } + log.Info("deployment already handled; not applying it again", + "outcome", prev.Outcome, "handled_revision", prev.Revision, "report_queued", queued) + notify(a.flushCh) + return true +} + // apply runs the pipeline of docs/adr/0021: fetch, verify, store, Validate, // Activate, Health, rollback. Shutdown aborts it until Activate starts; from // then on it is completed (bounded by the configured timeouts) so that the @@ -488,6 +532,9 @@ func (a *Agent) finish(log *slog.Logger, r DeploymentReport, update func(*State) last := r a.st.LastResult = &last a.st.PendingReports = append(a.st.PendingReports, r) + if r.Trigger == TriggerPush { + a.st.rememberDeployment(r) + } a.st.UpdatedAt = time.Now().UTC() err := a.store.SaveState(a.st) a.mu.Unlock() @@ -535,12 +582,17 @@ func (a *Agent) flushReports(ctx context.Context) { rctx, cancel := context.WithTimeout(ctx, reportTimeout) err := a.source.ReportDeployment(rctx, r) cancel() - if err != nil { + switch { + case errors.Is(err, ErrReportRefused): + a.log.Warn("deployment report refused; dropping it", "revision", r.Revision, + "deployment_id", r.DeploymentID, "error", err) + case err != nil: a.log.Debug("cannot deliver deployment report; will retry", "revision", r.Revision, "deployment_id", r.DeploymentID, "error", err) return + default: + a.log.Info("deployment reported", "revision", r.Revision, "deployment_id", r.DeploymentID, "outcome", r.Outcome) } - a.log.Info("deployment reported", "revision", r.Revision, "deployment_id", r.DeploymentID, "outcome", r.Outcome) a.mu.Lock() if i := slices.IndexFunc(a.st.PendingReports, func(p DeploymentReport) bool { return reflect.DeepEqual(p, r) }); i >= 0 { a.st.PendingReports = slices.Delete(a.st.PendingReports, i, i+1) diff --git a/agent/core/agent_test.go b/agent/core/agent_test.go index 0e2db44..b9b3a64 100644 --- a/agent/core/agent_test.go +++ b/agent/core/agent_test.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "slices" + "strconv" "strings" "testing" "time" @@ -385,7 +386,7 @@ func TestPendingReportsBounded(t *testing.T) { f := newFixture(t, KindDNS) f.src.setReportErr(errors.New("down")) for i := 0; i < maxPendingReports+5; i++ { - f.push(1, "dep") // fetch fails: rejected report each time + f.push(1, "dep-"+strconv.Itoa(i)) // fetch fails: rejected report each time } st, err := f.store.LoadState() if err != nil { diff --git a/agent/core/helpers_test.go b/agent/core/helpers_test.go index a8dfe31..816513e 100644 --- a/agent/core/helpers_test.go +++ b/agent/core/helpers_test.go @@ -456,6 +456,14 @@ type fakeServer struct { polls int commands chan *agentv1.Command enrollments int + // pushDeployment is announced by GetDesiredState until a report with + // this deployment ID arrives (open push, ADR 0022). + pushDeployment string + // reportCode makes ReportDeployment fail with this code if set. + reportCode codes.Code + // kick ends the current Connect stream (as the hub does on a mode + // change). + kick chan struct{} } func newFakeServer(t *testing.T, role Kind) *fakeServer { @@ -467,7 +475,7 @@ func newFakeServer(t *testing.T, role Kind) *fakeServer { enrollValidity: [2]time.Duration{-time.Minute, 14 * 24 * time.Hour}, mode: agentv1.OperatingMode_OPERATING_MODE_AUTO_PULL, pullInterval: 50 * time.Millisecond, heartbeat: 50 * time.Millisecond, - commands: make(chan *agentv1.Command, 10), + commands: make(chan *agentv1.Command, 10), kick: make(chan struct{}, 1), } } @@ -632,6 +640,8 @@ func (f *fakeServer) Connect(stream agentv1.AgentService_ConnectServer) error { return err case <-stream.Context().Done(): return nil + case <-f.kick: + return nil case c := <-f.commands: if err := stream.Send(&agentv1.ConnectResponse{Message: &agentv1.ConnectResponse_Command{Command: c}}); err != nil { return err @@ -647,7 +657,10 @@ func (f *fakeServer) GetDesiredState(ctx context.Context, _ *agentv1.GetDesiredS f.mu.Lock() defer f.mu.Unlock() f.polls++ - return &agentv1.GetDesiredStateResponse{DesiredRevision: f.desired, Mode: f.mode, PullInterval: durationpb.New(f.pullInterval)}, nil + return &agentv1.GetDesiredStateResponse{ + DesiredRevision: f.desired, Mode: f.mode, PullInterval: durationpb.New(f.pullInterval), + DeploymentId: f.pushDeployment, + }, nil } func (f *fakeServer) FetchBundle(ctx context.Context, req *agentv1.FetchBundleRequest) (*agentv1.FetchBundleResponse, error) { @@ -675,7 +688,13 @@ func (f *fakeServer) ReportDeployment(ctx context.Context, req *agentv1.ReportDe } f.mu.Lock() defer f.mu.Unlock() + if f.reportCode != codes.OK { + return nil, status.Error(f.reportCode, "report refused") + } f.reports = append(f.reports, req) + if req.GetDeploymentId() != "" && req.GetDeploymentId() == f.pushDeployment { + f.pushDeployment = "" + } return &agentv1.ReportDeploymentResponse{}, nil } diff --git a/agent/core/push_test.go b/agent/core/push_test.go new file mode 100644 index 0000000..d3de72c --- /dev/null +++ b/agent/core/push_test.go @@ -0,0 +1,366 @@ +// SPDX-License-Identifier: Apache-2.0 + +package core + +import ( + "context" + "errors" + "fmt" + "slices" + "strings" + "testing" + "time" + + "google.golang.org/grpc/codes" + + agentv1 "git.beugsiud.de/beug.k/DDI/proto/gen/ddi/agent/v1" +) + +// Tests for push deployments of docs/adr/0022-ausrollen-pull-push-m4.md. + +// countCalls counts the role calls that start with prefix. +func (r *fakeRole) countCalls(prefix string) int { + n := 0 + for _, c := range r.callLog() { + if strings.HasPrefix(c, prefix) { + n++ + } + } + return n +} + +// activations counts the activations of a revision. +func (r *fakeRole) activations(rev uint64) int { + return r.countCalls(fmt.Sprintf("activate %d ", rev)) +} + +// validations counts the validations of a revision. +func (r *fakeRole) validations(rev uint64) int { + return r.countCalls(fmt.Sprintf("validate %d", rev)) +} + +func (f *fixture) pending() []DeploymentReport { + f.agent.mu.Lock() + defer f.agent.mu.Unlock() + return slices.Clone(f.agent.st.PendingReports) +} + +func TestPushAppliedOncePerDeployment(t *testing.T) { + ctx := context.Background() + f := newFixture(t, KindDNS) + f.addBundle(t, 1) + f.push(1, "dep-1") + if n := f.role.activations(1); n != 1 { + t.Fatalf("activations of revision 1 = %d", n) + } + + // The server resends the open push (reconnect, polling) while the + // report is still queued: not applied again, the queued report is + // delivered once. + f.push(1, "dep-1") + if n := f.role.activations(1); n != 1 { + t.Errorf("resent push applied again (%d activations)", n) + } + if p := f.pending(); len(p) != 1 || p[0].DeploymentID != "dep-1" || p[0].Outcome != OutcomeSucceeded { + t.Fatalf("pending reports = %+v", p) + } + f.agent.flushReports(ctx) + if r := f.src.reported(); len(r) != 1 { + t.Fatalf("reports = %+v", r) + } + + // Resent after the report was delivered (the server did not record + // it): the remembered report is delivered again. + f.push(1, "dep-1") + if n := f.role.activations(1); n != 1 { + t.Errorf("resent push applied again (%d activations)", n) + } + f.agent.flushReports(ctx) + r := f.src.reported() + if len(r) != 2 || r[1].DeploymentID != "dep-1" || r[1].Outcome != OutcomeSucceeded || + !r[1].StartedAt.Equal(r[0].StartedAt) || r[1].Trigger != TriggerPush { + t.Fatalf("reports = %+v", r) + } + + // A new deployment of the same revision is applied again. + f.push(1, "dep-2") + if n := f.role.activations(1); n != 2 { + t.Errorf("activations of revision 1 = %d, want 2", n) + } + + // Pushes without a deployment ID are never deduplicated (M3 behavior). + f.push(1, "") + f.push(1, "") + if n := f.role.activations(1); n != 4 { + t.Errorf("activations of revision 1 = %d, want 4", n) + } +} + +func TestRejectedPushNotRetriedForSameDeployment(t *testing.T) { + f := newFixture(t, KindDNS) + f.addBundle(t, 1) + f.pull(1) + f.addBundle(t, 2) + f.role.mu.Lock() + f.role.validateErr[2] = errors.New("bad zone") + f.role.mu.Unlock() + f.push(2, "dep-3") + if r := f.lastResult(t); r.Outcome != OutcomeRejected || r.DeploymentID != "dep-3" { + t.Fatalf("result = %+v", r) + } + f.role.mu.Lock() + delete(f.role.validateErr, 2) + f.role.mu.Unlock() + + f.push(2, "dep-3") + if n := f.role.validations(2); n != 1 { + t.Errorf("validations of revision 2 = %d, want 1", n) + } + if st := f.agent.state(); st.Applied != 1 { + t.Errorf("applied = %d", st.Applied) + } + // A new push of the same revision retries it. + f.push(2, "dep-4") + if r := f.lastResult(t); r.Outcome != OutcomeSucceeded || r.DeploymentID != "dep-4" { + t.Errorf("result = %+v", r) + } +} + +func TestHandledDeploymentsPersistedAndBounded(t *testing.T) { + f := newFixture(t, KindDNS) + f.addBundle(t, 1) + for i := range maxHandledDeployments + 5 { + f.push(1, fmt.Sprintf("dep-%d", i)) + } + st, err := f.store.LoadState() + if err != nil { + t.Fatal(err) + } + h := st.HandledDeployments + if len(h) != maxHandledDeployments || h[0].DeploymentID != "dep-5" || + h[len(h)-1].DeploymentID != fmt.Sprintf("dep-%d", maxHandledDeployments+4) { + t.Fatalf("handled deployments = %+v", h) + } + + // After a restart, resent pushes are still recognized; forgotten ones + // are applied again. + g := newFixtureWithStore(t, KindDNS, f.store, f.signer) + g.addBundle(t, 1) + g.push(1, fmt.Sprintf("dep-%d", maxHandledDeployments+4)) + if n := g.role.activations(1); n != 0 { + t.Errorf("remembered deployment applied again after restart (%d activations)", n) + } + g.push(1, "dep-0") + if n := g.role.activations(1); n != 1 { + t.Errorf("forgotten deployment: %d activations, want 1", n) + } +} + +func TestRememberDeployment(t *testing.T) { + var st State + st.rememberDeployment(DeploymentReport{Revision: 1}) // pulls are not remembered + st.rememberDeployment(DeploymentReport{DeploymentID: "a", Revision: 1, Message: strings.Repeat("x", 5000)}) + st.rememberDeployment(DeploymentReport{DeploymentID: "b", Revision: 2}) + st.rememberDeployment(DeploymentReport{DeploymentID: "a", Revision: 3}) + if len(st.HandledDeployments) != 2 || st.HandledDeployments[0].DeploymentID != "b" || + st.HandledDeployments[1].Revision != 3 { + t.Fatalf("handled = %+v", st.HandledDeployments) + } + st.rememberDeployment(DeploymentReport{DeploymentID: "c", Message: strings.Repeat("y", 5000)}) + if r, ok := st.handledDeployment("c"); !ok || len(r.Message) > handledMessageLen { + t.Errorf("handled c = %v, message length %d", ok, len(r.Message)) + } + if _, ok := st.handledDeployment(""); ok { + t.Error("empty deployment ID found") + } + if _, ok := st.handledDeployment("x"); ok { + t.Error("unknown deployment ID found") + } +} + +func TestEnqueueDeduplicatesPushes(t *testing.T) { + f := newFixture(t, KindDNS) + a := f.agent + a.enqueue(applyRequest{revision: 3, trigger: TriggerPush, deploymentID: "d"}) + a.enqueue(applyRequest{revision: 3, trigger: TriggerPush, deploymentID: "d"}) + a.enqueue(applyRequest{revision: 3, trigger: TriggerPush, deploymentID: "e"}) + a.enqueue(applyRequest{revision: 3, trigger: TriggerPush}) + a.enqueue(applyRequest{revision: 3, trigger: TriggerPush}) + var got []string + for { + r, ok := a.dequeue() + if !ok { + break + } + got = append(got, r.deploymentID) + } + if want := []string{"d", "e", "", ""}; !slices.Equal(got, want) { + t.Errorf("queued deployments = %q, want %q", got, want) + } +} + +func TestRefusedReportIsDropped(t *testing.T) { + f := newFixture(t, KindDNS) + f.addBundle(t, 1) + f.addBundle(t, 2) + f.pull(1) + f.push(2, "dep-gone") + f.src.setReportErr(fmt.Errorf("%w: deployment closed", ErrReportRefused)) + f.agent.flushReports(context.Background()) + if p := f.pending(); len(p) != 0 { + t.Errorf("refused reports still queued: %+v", p) + } +} + +func TestServerSourceReportRefused(t *testing.T) { + srv := newFakeServer(t, KindDNS) + addr := srv.start(t) + store, id := enrollWith(t, srv, addr) + src := NewServerSource(store, id, "h", testLogger()) + defer src.closeConn() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + rep := DeploymentReport{DeploymentID: "d", Revision: 1, Trigger: TriggerPush, Outcome: OutcomeSucceeded, + ActiveRevision: 1, StartedAt: time.Now(), FinishedAt: time.Now()} + if err := src.ReportDeployment(ctx, rep); err != nil { + t.Fatalf("report = %v", err) + } + srv.with(func(f *fakeServer) { f.reportCode = codes.FailedPrecondition }) + if err := src.ReportDeployment(ctx, rep); !errors.Is(err, ErrReportRefused) { + t.Errorf("refused report = %v", err) + } + srv.with(func(f *fakeServer) { f.reportCode = codes.Unavailable }) + if err := src.ReportDeployment(ctx, rep); err == nil || errors.Is(err, ErrReportRefused) { + t.Errorf("unavailable server = %v", err) + } +} + +func TestServerSourcePollingPush(t *testing.T) { + srv := newFakeServer(t, KindDNS) + addr := srv.start(t) + store, id := enrollWith(t, srv, addr) + srv.with(func(f *fakeServer) { + f.desired = 1 + f.mode = agentv1.OperatingMode_OPERATING_MODE_MANUAL + f.streamDisabled = true + }) + role, _ := runAgentAgainst(t, store, id, KindDNS) + polls := func() int { + n := 0 + srv.with(func(f *fakeServer) { n = f.polls }) + return n + } + waitFor(t, "polls", func() bool { return polls() >= 3 }) + if n := len(serverReports(srv)); n != 0 { + t.Fatalf("manual mode applied the desired revision (%d reports)", n) + } + + // An open push is announced by polling and applied in manual mode. + srv.with(func(f *fakeServer) { + f.desired = 2 + f.pushDeployment = "dep-poll" + }) + waitFor(t, "report of the push", func() bool { return len(serverReports(srv)) >= 1 }) + r := serverReports(srv)[0] + if r.GetRevision() != 2 || r.GetDeploymentId() != "dep-poll" || + r.GetTrigger() != agentv1.DeploymentTrigger_DEPLOYMENT_TRIGGER_PUSH || + r.GetOutcome() != agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_SUCCEEDED { + t.Fatalf("report = %v", r) + } + after := polls() + waitFor(t, "more polls", func() bool { return polls() >= after+3 }) + if n := role.activations(2); n != 1 { + t.Errorf("activations of revision 2 = %d, want 1", n) + } + // A poll racing with the report may cause the same report to be sent + // again, never a second apply. + for _, rr := range serverReports(srv) { + if rr.GetDeploymentId() != "dep-poll" || !rr.GetStartedAt().AsTime().Equal(r.GetStartedAt().AsTime()) { + t.Errorf("unexpected report %v", rr) + } + } +} + +func TestServerSourceModeChangeAndResentPush(t *testing.T) { + srv := newFakeServer(t, KindDNS) + addr := srv.start(t) + store, id := enrollWith(t, srv, addr) + srv.with(func(f *fakeServer) { + f.desired = 1 + f.mode = agentv1.OperatingMode_OPERATING_MODE_MANUAL + }) + role, _ := runAgentAgainst(t, store, id, KindDNS) + counts := func() (hellos, heartbeats, acks int) { + srv.with(func(f *fakeServer) { hellos, heartbeats, acks = len(f.hellos), f.heartbeats, len(f.acks) }) + return + } + waitFor(t, "heartbeats in manual mode", func() bool { _, hb, _ := counts(); return hb >= 3 }) + if n := len(serverReports(srv)); n != 0 { + t.Fatalf("manual mode applied the desired revision (%d reports)", n) + } + + // Switching to auto_pull: the server closes the stream and the new + // Welcome carries the mode. + srv.with(func(f *fakeServer) { f.mode = agentv1.OperatingMode_OPERATING_MODE_AUTO_PULL }) + srv.kick <- struct{}{} + waitFor(t, "report of revision 1", func() bool { return len(serverReports(srv)) == 1 }) + if r := serverReports(srv)[0]; r.GetRevision() != 1 || r.GetTrigger() != agentv1.DeploymentTrigger_DEPLOYMENT_TRIGGER_PULL { + t.Fatalf("report = %v", r) + } + if h, _, _ := counts(); h < 2 { + t.Errorf("hellos = %d, want a reconnect", h) + } + + // Back to manual with a newer revision: it is not applied. + srv.with(func(f *fakeServer) { + f.mode = agentv1.OperatingMode_OPERATING_MODE_MANUAL + f.desired = 2 + }) + srv.kick <- struct{}{} + waitFor(t, "third hello", func() bool { h, _, _ := counts(); return h >= 3 }) + _, hb, _ := counts() + waitFor(t, "heartbeats after the mode change", func() bool { _, n, _ := counts(); return n >= hb+3 }) + if n := role.activations(2); n != 0 || len(serverReports(srv)) != 1 { + t.Fatalf("manual mode applied revision 2 (%d activations)", n) + } + + // The explicit push applies it. + push := func(cmdID string) *agentv1.Command { + return &agentv1.Command{CommandId: cmdID, Action: &agentv1.Command_ApplyRevision{ + ApplyRevision: &agentv1.ApplyRevisionCommand{Revision: 2, DeploymentId: "dep-2"}, + }} + } + srv.commands <- push("c1") + waitFor(t, "report of the push", func() bool { return len(serverReports(srv)) == 2 }) + first := serverReports(srv)[1] + if first.GetRevision() != 2 || first.GetDeploymentId() != "dep-2" || + first.GetTrigger() != agentv1.DeploymentTrigger_DEPLOYMENT_TRIGGER_PUSH || + first.GetOutcome() != agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_SUCCEEDED { + t.Fatalf("report = %v", first) + } + + // The server resends the push (as after a reconnect) after the agent + // has dropped the delivered report from its queue: acknowledged, not + // applied again, the remembered report is delivered again. + waitFor(t, "empty report queue", func() bool { + st, err := store.LoadState() + return err == nil && len(st.PendingReports) == 0 + }) + srv.commands <- push("c2") + waitFor(t, "second ack and report", func() bool { + _, _, acks := counts() + return acks == 2 && len(serverReports(srv)) == 3 + }) + if n := role.activations(2); n != 1 { + t.Errorf("activations of revision 2 = %d, want 1", n) + } + again := serverReports(srv)[2] + if again.GetDeploymentId() != "dep-2" || !again.GetStartedAt().AsTime().Equal(first.GetStartedAt().AsTime()) { + t.Errorf("repeated report = %v", again) + } + srv.with(func(f *fakeServer) { + if !f.acks[1].GetAccepted() || f.acks[1].GetCommandId() != "c2" { + t.Errorf("acks = %v", f.acks) + } + }) +} diff --git a/agent/core/server_source.go b/agent/core/server_source.go index 95c600a..d058e13 100644 --- a/agent/core/server_source.go +++ b/agent/core/server_source.go @@ -14,8 +14,10 @@ import ( "time" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" "google.golang.org/grpc/keepalive" + "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/timestamppb" "git.beugsiud.de/beug.k/DDI/internal/agentpki" @@ -353,8 +355,15 @@ func (s *ServerSource) poll(ctx context.Context, h SourceHandler) error { return err } s.setIntervals(resp.GetPullInterval().AsDuration(), 0) - s.log.Debug("desired state polled", "desired_revision", resp.GetDesiredRevision(), "mode", resp.GetMode().String()) - h.Desired(resp.GetDesiredRevision(), resp.GetMode() != agentv1.OperatingMode_OPERATING_MODE_MANUAL) + rev, dep := resp.GetDesiredRevision(), resp.GetDeploymentId() + s.log.Debug("desired state polled", "desired_revision", rev, "mode", resp.GetMode().String(), "deployment_id", dep) + if dep != "" && rev != 0 { + // An open push: applied in every mode (docs/adr/0022-ausrollen-pull-push-m4.md). + s.log.Info("open push received by polling", "revision", rev, "deployment_id", dep) + h.Push(rev, dep) + return nil + } + h.Desired(rev, resp.GetMode() != agentv1.OperatingMode_OPERATING_MODE_MANUAL) return nil } @@ -477,9 +486,23 @@ func (s *ServerSource) ReportDeployment(ctx context.Context, r DeploymentReport) Outcome: outcomeToProto(r.Outcome), Message: r.Message, ActiveRevision: r.ActiveRevision, StartedAt: timestamppb.New(r.StartedAt), FinishedAt: timestamppb.New(r.FinishedAt), }) + if refusedReportCodes[status.Code(err)] { + return fmt.Errorf("%w: %w", ErrReportRefused, err) + } return err } +// refusedReportCodes are the answers to ReportDeployment that a retry +// cannot change: the server considers the report invalid or does not +// expect it (any more). +var refusedReportCodes = map[codes.Code]bool{ + codes.InvalidArgument: true, + codes.NotFound: true, + codes.AlreadyExists: true, + codes.FailedPrecondition: true, + codes.OutOfRange: true, +} + // ReportLeases implements Source. func (s *ServerSource) ReportLeases(ctx context.Context, b LeaseBatch) error { cl, err := s.client() diff --git a/agent/core/source.go b/agent/core/source.go index b089497..af751af 100644 --- a/agent/core/source.go +++ b/agent/core/source.go @@ -4,9 +4,15 @@ package core import ( "context" + "errors" "time" ) +// ErrReportRefused marks a deployment report that the source refused for +// good (invalid, or for a deployment it no longer expects). Retrying it +// would block the reports queued behind it. +var ErrReportRefused = errors.New("deployment report refused") + // Source is where the agent learns its desired revision, fetches bundles and // reports results (docs/adr/0010-agent-binaries-pro-rolle.md). M3 provides // ServerSource (gRPC); a FileSource for standalone operation (M7) implements @@ -18,7 +24,9 @@ type Source interface { // FetchBundle returns the signed bundle of a revision. The caller // verifies it. FetchBundle(ctx context.Context, revision uint64) (*Bundle, error) - // ReportDeployment delivers an apply outcome. + // ReportDeployment delivers an apply outcome. An error wrapping + // ErrReportRefused means that the report will never be accepted; the + // agent then drops it instead of retrying. ReportDeployment(ctx context.Context, r DeploymentReport) error // ReportLeases delivers a batch of lease observations. ReportLeases(ctx context.Context, b LeaseBatch) error @@ -34,7 +42,9 @@ type SourceHandler interface { // false in manual mode, where only explicit pushes are applied. Desired(revision uint64, autoApply bool) // Push asks for an explicit apply of a revision. Unlike Desired, it - // also retries a revision that failed before. + // also retries a revision that failed before. A push whose deployment + // ID was handled before is not applied again; its report is delivered + // (again) instead. Push(revision uint64, deploymentID string) // Status returns what the agent reports in Hello and heartbeats. Status(ctx context.Context) AgentStatus diff --git a/agent/core/store.go b/agent/core/store.go index 9126a69..2498a6b 100644 --- a/agent/core/store.go +++ b/agent/core/store.go @@ -38,6 +38,13 @@ const KeepRevisions = 10 // maxPendingReports bounds the queue of unsent deployment reports. const maxPendingReports = 100 +// maxHandledDeployments bounds the remembered push deployments, and +// handledMessageLen the message kept with each of them. +const ( + maxHandledDeployments = 20 + handledMessageLen = 1024 +) + // Store manages the agent's state directory. Writes are atomic: files are // written to a temporary file, synced and renamed; directories are built // under a temporary name and renamed into place. @@ -114,7 +121,41 @@ type State struct { LeaseSequence uint64 `json:"lease_sequence,omitempty"` // PendingReports are deployment reports not yet accepted by the server. PendingReports []DeploymentReport `json:"pending_reports,omitempty"` - UpdatedAt time.Time `json:"updated_at"` + // HandledDeployments are the reports of the most recent push + // deployments (oldest first, at most maxHandledDeployments). A push + // with one of these deployment IDs is not applied again: the server + // resends open pushes after a reconnect and announces them while + // polling until it has the report (docs/adr/0022-ausrollen-pull-push-m4.md). + HandledDeployments []DeploymentReport `json:"handled_deployments,omitempty"` + UpdatedAt time.Time `json:"updated_at"` +} + +// handledDeployment returns the remembered report of a push deployment. +func (st *State) handledDeployment(id string) (DeploymentReport, bool) { + if id == "" { + return DeploymentReport{}, false + } + i := slices.IndexFunc(st.HandledDeployments, func(d DeploymentReport) bool { return d.DeploymentID == id }) + if i < 0 { + return DeploymentReport{}, false + } + return st.HandledDeployments[i], true +} + +// rememberDeployment records the report of a finished push deployment. +func (st *State) rememberDeployment(r DeploymentReport) { + if r.DeploymentID == "" { + return + } + r.Message = truncate(r.Message, handledMessageLen) + handled := slices.DeleteFunc(slices.Clone(st.HandledDeployments), func(d DeploymentReport) bool { + return d.DeploymentID == r.DeploymentID + }) + handled = append(handled, r) + if n := len(handled); n > maxHandledDeployments { + handled = slices.Clone(handled[n-maxHandledDeployments:]) + } + st.HandledDeployments = handled } // LoadState reads state.json; a missing file yields the zero state. @@ -138,6 +179,9 @@ func (s *Store) SaveState(st *State) error { if n := len(st.PendingReports); n > maxPendingReports { st.PendingReports = slices.Clone(st.PendingReports[n-maxPendingReports:]) } + if n := len(st.HandledDeployments); n > maxHandledDeployments { + st.HandledDeployments = slices.Clone(st.HandledDeployments[n-maxHandledDeployments:]) + } data, err := json.MarshalIndent(st, "", " ") if err != nil { return fmt.Errorf("encode state: %w", err) From e7a0fbb12317118c06b6792de76fe9e34146e248 Mon Sep 17 00:00:00 2001 From: "beug.k@beugsiud.de" Date: Tue, 13 Oct 2026 20:11:27 +0200 Subject: [PATCH 2/2] test(e2e): pending changes, manual mode with push, diff, rollback, events New steps before the revoke step: /config/pending reports nothing after revision 2 and the new address afterwards; the DHCP agent switched to manual mode stays on revision 2 while the DNS agent reaches 3, and an explicit push of revision 3 brings it there with a succeeded push deployment; the diff of revision 3 contains the new host; rolling back to revision 1 creates revision 4 with SOA serial 4 and without the later names; /events announces a revision created while listening. Co-Authored-By: Claude Opus 5.5 --- test/e2e/run.sh | 70 +++++++++++++++++- test/e2e/server/api.sh | 158 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 225 insertions(+), 3 deletions(-) diff --git a/test/e2e/run.sh b/test/e2e/run.sh index 4e2c4c1..c945d3e 100755 --- a/test/e2e/run.sh +++ b/test/e2e/run.sh @@ -2,8 +2,10 @@ # SPDX-License-Identifier: Apache-2.0 # # End-to-end test of the M3 path: API change -> revision -> agent fetches, -# verifies and applies -> BIND/Kea serve it. Needs Docker with compose and -# the binaries from `make build` (BIN_DIR, default bin/). +# verifies and applies -> BIND/Kea serve it; and of M4: pending changes, +# manual mode with explicit push, diff, rollback and live events. Needs +# Docker with compose and the binaries from `make build` (BIN_DIR, default +# bin/). # # Usage: test/e2e/run.sh (make test-e2e) # Environment: BIN_DIR, KEEP=1 keeps the containers after the run. @@ -18,6 +20,10 @@ case "$bin_dir" in /*) ;; *) bin_dir="$root/$bin_dir" ;; esac dc() { docker compose -f "$here/compose.yaml" "$@"; } api() { dc exec -T server /e2e/api.sh "$@"; } step() { printf '\n=== %s\n' "$*"; } +fail() { + echo "$*" >&2 + exit 1 +} cleanup() { local rc=$? @@ -78,9 +84,14 @@ expect() { fi echo "ok $* -> $got" } +soa_serial() { + local soa + soa="$(dig_short e2e.test SOA)" + awk '{print $3}' <<<"$soa" +} expect "192.0.2.10" dig_short www.e2e.test A expect "www.e2e.test." dig_short -x 192.0.2.10 -expect "1" sh -c "docker compose -f '$here/compose.yaml' exec -T bind dig @127.0.0.1 e2e.test SOA +short | awk '{print \$3}'" +expect "1" soa_serial dc exec -T kea grep -q '"hw-address": "00:11:22:33:44:55"' /etc/kea/kea-dhcp4.conf echo "ok Kea configuration contains the reservation" @@ -103,6 +114,59 @@ api deployments | grep -c '"outcome": "succeeded"' | { echo "ok $n successful deployments" } +step "Pending changes are reported (M4)" +pending="$(api pending)" +read -r changed changes files <<<"$pending" +[[ "$changed" == false ]] || fail "pending changes right after revision 2: changed=$changed changes=$changes files=$files" +echo "ok nothing pending after revision 2" +api address 192.0.2.30 ftp +pending="$(api pending)" +read -r changed changes files <<<"$pending" +[[ "$changed" == true && "$changes" -ge 1 && "$files" -ge 1 ]] || + fail "new address not pending: changed=$changed changes=$changes files=$files" +echo "ok $changes pending change(s) in $files file(s)" + +step "Manual mode: the DHCP agent waits for an explicit push" +api mode dhcp manual +api wait-mode dhcp manual +rev="$(api revision "ftp server")" +[[ "$rev" == 3 ]] || fail "third revision is $rev" +api wait-agent dns 3 +expect "192.0.2.30" dig_short ftp.e2e.test A +sleep 5 +got="$(api agent-revision dhcp)" +[[ "$got" == 2 ]] || fail "DHCP agent in manual mode runs revision $got instead of 2" +echo "ok DHCP agent stays on revision 2" + +step "Pushing revision 3 to the DHCP agent" +dep="$(api deploy 3 dhcp)" +[[ "$dep" =~ ^[0-9]+$ ]] || fail "unexpected deployment ID '$dep'" +api wait-agent dhcp 3 +api wait-deployment "$dep" succeeded +echo "ok push deployment $dep succeeded" + +step "Diff of revision 3 shows the new host" +n="$(api diff-added 3 ftp)" +[[ "$n" -ge 1 ]] || fail "diff of revision 3 has no added line with ftp" +echo "ok $n added line(s) with ftp" + +step "Rollback to revision 1 creates revision 4" +rev="$(api rollback 1)" +[[ "$rev" == 4 ]] || fail "rollback created revision $rev" +api wait-agent dns 4 +expect "" dig_short mail.e2e.test A +expect "" dig_short ftp.e2e.test A +expect "" dig_short -x 192.0.2.20 +expect "192.0.2.10" dig_short www.e2e.test A +expect "4" soa_serial + +step "Live events (server-sent events)" +rev="$(api events-revision "after rollback")" +[[ "$rev" == 5 ]] || fail "revision during the event stream is $rev" +echo "ok event stream announced revision $rev" +api wait-agent dns 5 +expect "192.0.2.20" dig_short mail.e2e.test A + step "Revoking an agent disconnects it; BIND keeps serving" api revoke dns api wait-disconnected dns diff --git a/test/e2e/server/api.sh b/test/e2e/server/api.sh index 952d1f7..dd391cc 100755 --- a/test/e2e/server/api.sh +++ b/test/e2e/server/api.sh @@ -14,6 +14,17 @@ # deployments print the deployment list # revoke revoke the agent of a role # wait-disconnected wait until that agent is disconnected +# pending print " <#changes> <#files>" of /config/pending +# mode set the operating mode of the agent of a role +# wait-mode wait until that agent is connected in that mode +# wait-agent wait until that agent runs healthily +# agent-revision print the applied revision of that agent +# deploy push to the agent of a role, print the deployment ID +# wait-deployment wait until the push deployment has +# diff-added print the number of added diff lines of containing +# rollback roll back to , print the new revision number +# events-revision create a revision while listening to /events, +# print its number; fails without an "event: revision" set -eu base=http://127.0.0.1:8080/api/v1 @@ -43,6 +54,30 @@ login() { session() { login "$password"; } +# agent_id ROLE prints the ID of the (not revoked) agent of a role. +agent_id() { + id=$(req GET /agents | jq -r --arg role "$1" \ + '[.items[] | select(.role == $role and .revoked_at == null)][0].id // empty') + if [ -z "$id" ]; then + echo "no agent with role $1" >&2 + return 1 + fi + printf '%s\n' "$id" +} + +# agent_json ROLE prints the agent of a role. +agent_json() { + req GET /agents | jq --arg role "$1" '[.items[] | select(.role == $role and .revoked_at == null)][0]' +} + +# dump prints agents and recent deployments for diagnosis. +dump() { + echo "--- agents" >&2 + req GET /agents | jq . >&2 || true + echo "--- deployments" >&2 + req GET '/deployments?limit=20' | jq . >&2 || true +} + case "$1" in setup) login "$2" @@ -111,6 +146,129 @@ case "$1" in echo "agent $2 is still connected" >&2 exit 1 ;; + pending) + session + req GET /config/pending >/tmp/pending.json + # Summary for the log, result on stdout. + jq -c '{latest_revision, changed, files: [.files[] | "\(.status) \(.path)"], + changes: [.changes[] | "\(.action) \(.object_type) \(.object_name)"]}' /tmp/pending.json >&2 + jq -r '"\(.changed) \(.changes | length) \(.files | length)"' /tmp/pending.json + ;; + mode) + session + id=$(agent_id "$2") + req PATCH "/agents/$id" "{\"mode\":\"$3\"}" | jq -e --arg mode "$3" '.mode == $mode' >/dev/null + ;; + wait-mode) + session + i=0 + while [ "$i" -lt 30 ]; do + ok=$(agent_json "$2" | jq --arg mode "$3" '.mode == $mode and .connected') + if [ "$ok" = "true" ]; then + # The server reconnects the agent to tell it the new mode; give + # the new stream time to settle. + sleep 3 + exit 0 + fi + i=$((i + 1)) + sleep 1 + done + echo "agent $2 is not connected in mode $3" >&2 + dump + exit 1 + ;; + wait-agent) + session + i=0 + while [ "$i" -lt 90 ]; do + ok=$(agent_json "$2" | jq --argjson rev "$3" '.applied_revision == $rev and .health.state == "ok"') + [ "$ok" = "true" ] && exit 0 + i=$((i + 1)) + sleep 1 + done + echo "agent $2 did not reach revision $3" >&2 + dump + exit 1 + ;; + agent-revision) + session + agent_json "$2" | jq -r .applied_revision + ;; + deploy) + session + id=$(agent_id "$3") + req POST "/revisions/$2/deploy" "{\"agent_ids\":[\"$id\"]}" >/tmp/deploy.json + jq -c . /tmp/deploy.json >&2 + jq -e --arg id "$id" '.items | length == 1 and .[0].agent_id == $id' /tmp/deploy.json >/dev/null || { + echo "unexpected deploy response" >&2 + exit 1 + } + jq -r '.items[0].deployment_id' /tmp/deploy.json + ;; + wait-deployment) + session + i=0 + while [ "$i" -lt 60 ]; do + d=$(req GET '/deployments?limit=100' | jq -c --argjson id "$2" '.items[] | select(.id == $id)') + if [ -n "$d" ]; then + outcome=$(printf '%s\n' "$d" | jq -r .outcome) + trigger=$(printf '%s\n' "$d" | jq -r .trigger) + if [ "$outcome" = "$3" ] && [ "$trigger" = "push" ]; then + printf '%s\n' "$d" >&2 + exit 0 + fi + case "$outcome" in + pending) ;; + *) + echo "deployment $2 ended with $outcome ($trigger): $d" >&2 + dump + exit 1 + ;; + esac + fi + i=$((i + 1)) + sleep 1 + done + echo "deployment $2 did not reach $3" >&2 + dump + exit 1 + ;; + diff-added) + session + req GET "/revisions/$2/diff" >/tmp/diff.json + n=$(jq --arg t "$3" '[.files[].diff | split("\n")[] | select(startswith("+") and contains($t))] | length' /tmp/diff.json) + if [ "$n" -eq 0 ]; then + echo "no added line with '$3' in the diff of revision $2:" >&2 + jq -r '.files[] | "--- \(.role) \(.path) (\(.status))\n\(.diff)"' /tmp/diff.json >&2 + fi + printf '%s\n' "$n" + ;; + rollback) + session + req POST "/revisions/$2/rollback" "{\"comment\":\"e2e rollback to $2\"}" | jq -r .number + ;; + events-revision) + session + out=/tmp/events.out + rm -f "$out" + curl -sSN --max-time 10 -b "$jar" -H 'Accept: text/event-stream' "$base/events" >"$out" 2>/tmp/events.err & + pid=$! + sleep 2 + n=$(req POST /revisions "{\"comment\":\"$2\"}" | jq -r .number) + i=0 + while [ "$i" -lt 8 ] && ! grep -q '^event: revision' "$out"; do + i=$((i + 1)) + sleep 1 + done + kill "$pid" 2>/dev/null || true + wait "$pid" 2>/dev/null || true + if ! grep -q '^event: revision' "$out"; then + echo "no revision event received; captured stream:" >&2 + cat "$out" /tmp/events.err >&2 + exit 1 + fi + printf '%s\n' "$n" + ;; *) echo "unknown step $1" >&2 exit 2