Merge branch 'worktree-agent-a8916aaf27575ea1e'

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)

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 {

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
 }
 

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)
+		}
+	})
+}

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()

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

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)

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

test/e2e/server/api.sh

@@ -14,6 +14,17 @@
 #   deployments                  print the deployment list
 #   revoke <role>                revoke the agent of a role
 #   wait-disconnected <role>     wait until that agent is disconnected
+#   pending                      print "<changed> <#changes> <#files>" of /config/pending
+#   mode <role> <mode>           set the operating mode of the agent of a role
+#   wait-mode <role> <mode>      wait until that agent is connected in that mode
+#   wait-agent <role> <rev>      wait until that agent runs <rev> healthily
+#   agent-revision <role>        print the applied revision of that agent
+#   deploy <rev> <role>          push <rev> to the agent of a role, print the deployment ID
+#   wait-deployment <id> <outcome>  wait until the push deployment <id> has <outcome>
+#   diff-added <rev> <text>      print the number of added diff lines of <rev> containing <text>
+#   rollback <rev>               roll back to <rev>, print the new revision number
+#   events-revision <comment>    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