Merge pull request 'Accept reverted/failed device status reports' (#26) from benvin/status-reverted into main

Reviewed-on: #26

.woodpecker/build.yaml

@@ -3,9 +3,9 @@ when:
 
 steps:
   - name: docker-build
-    image: woodpeckerci/plugin-docker-buildx
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
     settings:
-      repo: git.unkin.net/unkin/tomswallapi
+      repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/tomswallapi
       dry_run: true
     backend_options:
       kubernetes:

.woodpecker/docker.yaml

@@ -4,15 +4,15 @@ when:
 
 steps:
   - name: docker-tomswallapi
-    image: woodpeckerci/plugin-docker-buildx
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
     settings:
-      registry: git.unkin.net
+      registry: artifactapi.k8s.syd1.au.unkin.net
-      repo: git.unkin.net/unkin/tomswallapi
+      repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/tomswallapi
+      buildkit_config: |
+        [registry."artifactapi.k8s.syd1.au.unkin.net"]
+          ca = ["/etc/docker/certs.d/artifactapi.k8s.syd1.au.unkin.net/ca.crt"]
       build_args:
         VERSION: ${CI_COMMIT_TAG}
-      username: droneci
-      password:
-        from_secret: DRONECI_PASSWORD
       tags:
         - ${CI_COMMIT_TAG}
         - latest

.woodpecker/pre-commit.yaml

@@ -3,9 +3,21 @@ when:
 
 steps:
   - name: pre-commit
-    image: git.unkin.net/unkin/almalinux9-gobuilder:20260606
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
     commands:
       - uvx pre-commit run --all-files
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-tomswallapi
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     backend_options:
       kubernetes:
         serviceAccountName: default

.woodpecker/test.yaml

@@ -3,9 +3,21 @@ when:
 
 steps:
   - name: lint
-    image: golang:1.25
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
     commands:
       - make lint
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-tomswallapi
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     backend_options:
       kubernetes:
         serviceAccountName: default
@@ -18,10 +30,22 @@ steps:
             cpu: 2
 
   - name: test
-    image: golang:1.25
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
     commands:
       # Container-backed DB tests self-skip when Docker is unavailable in CI.
       - make test-short
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-tomswallapi
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     backend_options:
       kubernetes:
         serviceAccountName: default

internal/database/migrations/0010_device_status.sql

@@ -0,0 +1,7 @@
+-- Last agent apply report per device. reported_generation keeps its meaning
+-- (last applied generation); these columns hold the latest report of any kind.
+ALTER TABLE devices
+    ADD COLUMN status TEXT NOT NULL DEFAULT '' CHECK (status IN ('', 'applied', 'reverted', 'failed')),
+    ADD COLUMN status_generation BIGINT NOT NULL DEFAULT 0,
+    ADD COLUMN status_error TEXT NOT NULL DEFAULT '',
+    ADD COLUMN status_at TIMESTAMPTZ;

internal/model/model.go

@@ -103,6 +103,48 @@ type Device struct {
 	// ReachablePrefixes is the device's FIB as last reported by its agent
 	// (server-managed). The compiler uses it to scope router enforcement.
 	ReachablePrefixes []string `json:"reachable_prefixes,omitempty"`
+
+	// AgentStatus is the agent's last apply report (server-managed).
+	AgentStatus *AgentStatus `json:"agent_status,omitempty"`
+}
+
+// ApplyStatus is the outcome an agent reports for a generation.
+type ApplyStatus string
+
+const (
+	StatusApplied  ApplyStatus = "applied"
+	StatusReverted ApplyStatus = "reverted"
+	StatusFailed   ApplyStatus = "failed"
+)
+
+// StatusReport is the agent's POST /devices/{name}/status body. An empty
+// Status (pre-status agents sending only generation) means applied.
+type StatusReport struct {
+	Status     ApplyStatus `json:"status,omitempty"`
+	Generation int64       `json:"generation"`
+	Error      string      `json:"error,omitempty"`
+}
+
+// Normalize defaults an empty Status to applied and rejects unknown values.
+func (r *StatusReport) Normalize() error {
+	switch r.Status {
+	case "":
+		r.Status = StatusApplied
+	case StatusApplied, StatusReverted, StatusFailed:
+	default:
+		return fmt.Errorf("invalid status %q: want applied, reverted or failed", r.Status)
+	}
+	return nil
+}
+
+// AgentStatus is a device's stored apply state. AppliedGeneration only advances
+// on applied reports; Status/Generation/Error/ReportedAt mirror the last report.
+type AgentStatus struct {
+	AppliedGeneration int64       `json:"applied_generation"`
+	Status            ApplyStatus `json:"status"`
+	Generation        int64       `json:"generation"`
+	Error             string      `json:"error,omitempty"`
+	ReportedAt        time.Time   `json:"reported_at"`
 }
 
 // Binding maps a global zone to one device's local interface(s).

internal/server/resources.go

@@ -383,13 +383,15 @@ func (s *Server) handleDeviceRoutes(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) handleDeviceStatus(w http.ResponseWriter, r *http.Request) {
-	var body struct {
+	var body model.StatusReport
-		Generation int64 `json:"generation"`
-	}
 	if !decode(w, r, &body) {
 		return
 	}
-	if err := s.store.RecordDeviceStatus(r.Context(), chi.URLParam(r, "name"), body.Generation); err != nil {
+	if err := body.Normalize(); err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	if err := s.store.RecordDeviceStatus(r.Context(), chi.URLParam(r, "name"), body); err != nil {
 		if errors.Is(err, store.ErrNotFound) {
 			writeError(w, http.StatusNotFound, "device not found")
 			return

internal/server/status_test.go

@@ -0,0 +1,45 @@
+package server
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"git.unkin.net/unkin/tomswallapi/internal/model"
+)
+
+func TestStatusReportNormalize(t *testing.T) {
+	tests := []struct {
+		in      model.ApplyStatus
+		want    model.ApplyStatus
+		wantErr bool
+	}{
+		{in: "", want: model.StatusApplied},
+		{in: model.StatusApplied, want: model.StatusApplied},
+		{in: model.StatusReverted, want: model.StatusReverted},
+		{in: model.StatusFailed, want: model.StatusFailed},
+		{in: "rolledback", wantErr: true},
+	}
+	for _, tt := range tests {
+		r := model.StatusReport{Status: tt.in, Generation: 3}
+		err := r.Normalize()
+		if (err != nil) != tt.wantErr || (!tt.wantErr && r.Status != tt.want) {
+			t.Errorf("Normalize(%q) = %q, %v", tt.in, r.Status, err)
+		}
+	}
+}
+
+func TestHandleDeviceStatusRejectsBadPayload(t *testing.T) {
+	s := &Server{} // nil store: a 400 must return before any DB access
+	for _, body := range []string{
+		`{"status":"rolledback","generation":3}`,
+		`{"generation":3,"bogus":1}`,
+	} {
+		rec := httptest.NewRecorder()
+		s.handleDeviceStatus(rec, httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body)))
+		if rec.Code != http.StatusBadRequest {
+			t.Errorf("%s: got %d, want 400", body, rec.Code)
+		}
+	}
+}

internal/store/store.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -279,11 +280,20 @@ func (s *Store) UpsertDevice(ctx context.Context, d model.Device) error {
 	})
 }
 
-// RecordDeviceStatus stores the generation an agent reports as applied.
+// RecordDeviceStatus stores an agent's apply report. reported_generation only
-func (s *Store) RecordDeviceStatus(ctx context.Context, name string, generation int64) error {
+// advances on applied; the last-status columns only take reports at or above
-	tag, err := s.pool.Exec(ctx,
+// the stored status generation, so late or replayed reports never clobber newer state.
-		`UPDATE devices SET reported_generation = $2, last_seen = now() WHERE name = $1`,
+func (s *Store) RecordDeviceStatus(ctx context.Context, name string, r model.StatusReport) error {
-		name, generation)
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE devices SET
+			reported_generation = CASE WHEN $2 = 'applied' THEN GREATEST(reported_generation, $3) ELSE reported_generation END,
+			status = CASE WHEN $3 >= status_generation THEN $2 ELSE status END,
+			status_error = CASE WHEN $3 >= status_generation THEN $4 ELSE status_error END,
+			status_at = CASE WHEN $3 >= status_generation THEN now() ELSE status_at END,
+			status_generation = GREATEST(status_generation, $3),
+			last_seen = now()
+		WHERE name = $1`,
+		name, string(r.Status), r.Generation, r.Error)
 	if err != nil {
 		return err
 	}
@@ -296,10 +306,14 @@ func (s *Store) RecordDeviceStatus(ctx context.Context, name string, generation
 func (s *Store) GetDevice(ctx context.Context, name string) (model.Device, error) {
 	var d model.Device
 	var resolver, settings, reachable []byte
+	var st model.AgentStatus
+	var statusAt *time.Time
 	err := s.pool.QueryRow(ctx,
-		`SELECT name, class, COALESCE(fabric, ''), resolver, settings, reachable_prefixes
+		`SELECT name, class, COALESCE(fabric, ''), resolver, settings, reachable_prefixes,
+		        reported_generation, status, status_generation, status_error, status_at
 		 FROM devices WHERE name = $1`, name,
-	).Scan(&d.Name, &d.Class, &d.Fabric, &resolver, &settings, &reachable)
+	).Scan(&d.Name, &d.Class, &d.Fabric, &resolver, &settings, &reachable,
+		&st.AppliedGeneration, &st.Status, &st.Generation, &st.Error, &statusAt)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return d, ErrNotFound
 	}
@@ -312,6 +326,10 @@ func (s *Store) GetDevice(ctx context.Context, name string) (model.Device, error
 	if err := json.Unmarshal(reachable, &d.ReachablePrefixes); err != nil {
 		return d, err
 	}
+	if statusAt != nil {
+		st.ReportedAt = *statusAt
+		d.AgentStatus = &st
+	}
 	return d, json.Unmarshal(settings, &d.Settings)
 }
 

internal/store/store_integration_test.go

@@ -149,3 +149,77 @@ func TestNATTierRoundTrip(t *testing.T) {
 		t.Errorf("expected nat rows to cascade-delete with device, got %d", len(nats))
 	}
 }
+
+func TestRecordDeviceStatus(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	if err := s.UpsertDevice(ctx, model.Device{Name: "fw1", Class: model.ClassFirewall}); err != nil {
+		t.Fatalf("upsert device: %v", err)
+	}
+	if d, _ := s.GetDevice(ctx, "fw1"); d.AgentStatus != nil {
+		t.Fatalf("unreported device should have no agent_status: %+v", d.AgentStatus)
+	}
+
+	report := func(r model.StatusReport) model.AgentStatus {
+		t.Helper()
+		if err := r.Normalize(); err != nil {
+			t.Fatalf("normalize: %v", err)
+		}
+		if err := s.RecordDeviceStatus(ctx, "fw1", r); err != nil {
+			t.Fatalf("record %+v: %v", r, err)
+		}
+		d, err := s.GetDevice(ctx, "fw1")
+		if err != nil || d.AgentStatus == nil {
+			t.Fatalf("get device: %v %+v", err, d.AgentStatus)
+		}
+		return *d.AgentStatus
+	}
+
+	// Old agent payload: generation only.
+	got := report(model.StatusReport{Generation: 5})
+	if got.AppliedGeneration != 5 || got.Status != model.StatusApplied || got.Generation != 5 || got.ReportedAt.IsZero() {
+		t.Errorf("old payload: %+v", got)
+	}
+
+	got = report(model.StatusReport{Status: model.StatusApplied, Generation: 6})
+	if got.AppliedGeneration != 6 || got.Status != model.StatusApplied {
+		t.Errorf("applied: %+v", got)
+	}
+
+	got = report(model.StatusReport{Status: model.StatusReverted, Generation: 7, Error: "api unreachable"})
+	if got.AppliedGeneration != 6 || got.Status != model.StatusReverted || got.Generation != 7 || got.Error != "api unreachable" {
+		t.Errorf("reverted must not advance applied generation: %+v", got)
+	}
+
+	got = report(model.StatusReport{Status: model.StatusFailed, Generation: 8, Error: "nft: syntax error"})
+	if got.AppliedGeneration != 6 || got.Status != model.StatusFailed || got.Generation != 8 || got.Error != "nft: syntax error" {
+		t.Errorf("failed must not advance applied generation: %+v", got)
+	}
+
+	got = report(model.StatusReport{Status: model.StatusApplied, Generation: 9})
+	if got.AppliedGeneration != 9 || got.Status != model.StatusApplied || got.Generation != 9 {
+		t.Errorf("applied 9: %+v", got)
+	}
+
+	got = report(model.StatusReport{Status: model.StatusApplied, Generation: 4})
+	if got.AppliedGeneration != 9 || got.Status != model.StatusApplied || got.Generation != 9 {
+		t.Errorf("older applied must not regress: %+v", got)
+	}
+
+	got = report(model.StatusReport{Status: model.StatusReverted, Generation: 8, Error: "stale"})
+	if got.AppliedGeneration != 9 || got.Status != model.StatusApplied || got.Generation != 9 || got.Error != "" {
+		t.Errorf("older reverted must not overwrite newer status: %+v", got)
+	}
+
+	got = report(model.StatusReport{Status: model.StatusReverted, Generation: 9, Error: "lost api"})
+	if got.AppliedGeneration != 9 || got.Status != model.StatusReverted || got.Generation != 9 || got.Error != "lost api" {
+		t.Errorf("same-generation reverted after applied must update: %+v", got)
+	}
+
+	if err := s.RecordDeviceStatus(ctx, "fw1", model.StatusReport{Status: "bogus", Generation: 10}); err == nil {
+		t.Error("db should reject unknown status")
+	}
+	if err := s.RecordDeviceStatus(ctx, "nope", model.StatusReport{Status: model.StatusApplied}); err != store.ErrNotFound {
+		t.Errorf("unknown device: want ErrNotFound, got %v", err)
+	}
+}