amadan

carlos / platform Public

Refuse incomplete instance record sets before steering

claude 0797d9beb863277024048eea415e2affc6b36e94 parent bbeefbc
5 files changed, +156 −28
  • internal/host/steering_loop.go +8 −5
  • internal/host/steering_loop_test.go +67 −22
  • internal/provision/maintenance_test.go +27 −0
  • internal/provision/reconcile.go +19 −1
  • internal/steering/completeness_control_test.go +35 −0

CI passed — run details

diff --git a/internal/host/steering_loop.go b/internal/host/steering_loop.go
index 6b2379d99..1f1750769 100644
--- a/internal/host/steering_loop.go
+++ b/internal/host/steering_loop.go
@@ -69,7 +69,7 @@ func loadEdgesFileBodies(ctx context.Context, s release.Store) ([][]byte, error)
}
// steeredRoutesFromRecords reads the deployment's steered hosts from the
-// BUCKET RECORDS — provision.ListAllRecords — and never from this box's
+// BUCKET RECORDS — provision.ListAllRecordsComplete — and never from this box's
// registry.
//
// The registry is the wrong source and the bug is not theoretical (review
@@ -95,9 +95,10 @@ func loadEdgesFileBodies(ctx context.Context, s release.Store) ([][]byte, error)
// nothing an operator would see is lost. Every other value — including
// invalid ones, wildcard hosts and pool-scoped rows — is passed through
// untouched so the fold's own loud, documented refusals still fire. This
-// loader must not become a second, quieter policy gate.
+// loader must not become a second, quieter policy gate. Malformed recognized
+// records refuse the whole pass because their steering membership is unknown.
func steeredRoutesFromRecords(ctx context.Context, s release.Store) ([]registry.Route, error) {
- recs, err := provision.ListAllRecords(ctx, s)
+ recs, err := provision.ListAllRecordsComplete(ctx, s)
if err != nil {
return nil, err
}
@@ -115,12 +116,14 @@ func steeredRoutesFromRecords(ctx context.Context, s release.Store) ([]registry.
// then converge. The registry is not consulted at all — see
// steeredRoutesFromRecords.
//
-// Both reads fail closed (see loadEdgesFileBodies and ListAllRecords) and
+// Both reads fail closed (see loadEdgesFileBodies and ListAllRecordsComplete) and
// BOTH must succeed before converge is called even once — a records read
// that failed is not "no steered hosts" any more than a failed edges listing
// is "no edges". converge is injected rather than called directly so this
// contract is testable without an AWS client: the real one is the closure
-// steeringLoop builds around steering.ConvergeDeployment.
+// steeringLoop builds around steering.ConvergeDeployment. Malformed edges-file
+// content is still skipped there, making that box's records deletable within
+// the shrink limits; this record-listing change does not close that gap.
func steeringPass(ctx context.Context, s release.Store, converge func(context.Context, []registry.Route, [][]byte) error) error {
edges, err := loadEdgesFileBodies(ctx, s)
if err != nil {
diff --git a/internal/host/steering_loop_test.go b/internal/host/steering_loop_test.go
index 3cf390200..f42fc910d 100644
--- a/internal/host/steering_loop_test.go
+++ b/internal/host/steering_loop_test.go
@@ -312,28 +312,73 @@ func TestSteeringPassFailsClosedOnRecordsRead(t *testing.T) {
}
}
-// The content-vs-visibility split, records half: a record that READS but does
-// not decode is skipped and logged, and the pass converges on the rest. One
-// malformed body must not blind the deployment to every other steered host.
-func TestSteeringPassSkipsUnparseableRecord(t *testing.T) {
- logs := captureLog(t)
- s, sqid := steeringWorld(t)
- if _, err := s.Put(context.Background(), provision.RecordKey(sqid, "app", "broken.example.com"), []byte("{not json"), ""); err != nil {
- t.Fatal(err)
- }
- rec := &convergeRecorder{}
-
- if err := steeringPass(context.Background(), s, rec.converge); err != nil {
- t.Fatalf("steeringPass: %v — an unparseable record is not a read failure", err)
- }
- if rec.calls != 1 {
- t.Fatalf("converge calls = %d, want 1", rec.calls)
- }
- if len(rec.routes) != 1 || rec.routes[0].Host != "app.example.com" {
- t.Fatalf("routes = %+v, want the healthy steered record still folded", rec.routes)
- }
- if !strings.Contains(logs(), "broken.example.com") {
- t.Fatalf("log = %q, want the malformed record named", logs())
+func TestSteeringPassRecordCompleteness(t *testing.T) {
+ for _, kind := range []string{"deleted", "maintenance_null", "invalid_json", "identity_mismatch"} {
+ t.Run(kind, func(t *testing.T) {
+ quietLog(t)
+ s := steeringTestStore(t, map[string][]byte{"control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`)})
+ sqid := seedAccount(t, s, 12, "Tito")
+ seedRecord(t, s, sqid, "app", "app.example.com", "latency", "")
+ seedRecord(t, s, sqid, "app", "other.example.com", "latency", "")
+ key := provision.RecordKey(sqid, "app", "app.example.com")
+ raw, etag, err := s.Get(context.Background(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ switch kind {
+ case "deleted":
+ if err := s.Delete(context.Background(), key); err != nil {
+ t.Fatal(err)
+ }
+ case "maintenance_null", "identity_mismatch":
+ var fields map[string]json.RawMessage
+ if err := json.Unmarshal(raw, &fields); err != nil {
+ t.Fatal(err)
+ }
+ if kind == "maintenance_null" {
+ fields["maintenance"] = json.RawMessage("null")
+ } else {
+ fields["host"] = json.RawMessage(`"other.example.com"`)
+ }
+ raw, err = json.Marshal(fields)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = s.Put(context.Background(), key, raw, etag); err != nil {
+ t.Fatal(err)
+ }
+ case "invalid_json":
+ if _, err = s.Put(context.Background(), key, []byte(`{`), etag); err != nil {
+ t.Fatal(err)
+ }
+ }
+ rec := &convergeRecorder{}
+ err = steeringPass(context.Background(), s, rec.converge)
+ malformed := kind != "deleted"
+ if malformed {
+ if err == nil || !strings.Contains(err.Error(), key) || rec.calls != 0 {
+ t.Fatalf("steeringPass error = %v, converge calls = %d, routes = %+v; want error and no convergence", err, rec.calls, rec.routes)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("steeringPass: %v", err)
+ }
+ if rec.calls != 1 {
+ t.Fatalf("converge calls = %d, want 1", rec.calls)
+ }
+ want := 1
+ if len(rec.routes) != want {
+ t.Fatalf("routes = %+v, want %d", rec.routes, want)
+ }
+ hosts := map[string]bool{}
+ for _, r := range rec.routes {
+ hosts[r.Host] = true
+ }
+ if !hosts["other.example.com"] || hosts["app.example.com"] {
+ t.Fatalf("routes = %+v", rec.routes)
+ }
+ })
}
}
diff --git a/internal/provision/maintenance_test.go b/internal/provision/maintenance_test.go
index 177fa8ffa..dda7fa02f 100644
--- a/internal/provision/maintenance_test.go
+++ b/internal/provision/maintenance_test.go
@@ -230,4 +230,31 @@ func TestMalformedProjectionReconcileAndLists(t *testing.T) {
if records, e := ListAllRecords(ctx, s); e != nil || len(records) != 0 {
t.Fatal("all-list semantics changed")
}
+ rec.Host = "a.go.tito.io"
+ if e := CreateRecord(ctx, s, rec); e != nil {
+ t.Fatal(e)
+ }
+ if records, e := ListAllRecordsComplete(ctx, s); e == nil || records != nil {
+ t.Fatal("complete list accepted invalid")
+ }
+}
+
+func TestCompleteListIgnoresUnrecognizedKeys(t *testing.T) {
+ s, sqid := setup(t)
+ rec := acmeRecord(sqid)
+ rec.Schema = 99
+ rec.Kind = "future-kind"
+ rec.Steering = "future-steering"
+ if e := CreateRecord(ctx, s, rec); e != nil {
+ t.Fatal(e)
+ }
+ for _, key := range []string{RecordsPrefix(sqid, rec.App) + "nested/bad.json", RecordKey("unknown", rec.App, "bad.example.com")} {
+ if _, e := s.Put(ctx, key, []byte("{not json"), ""); e != nil {
+ t.Fatal(e)
+ }
+ }
+ records, e := ListAllRecordsComplete(ctx, s)
+ if e != nil || len(records) != 1 || records[0].Host != rec.Host || records[0].Schema != 99 || records[0].Kind != rec.Kind || records[0].Steering != rec.Steering {
+ t.Fatalf("records = %+v, error = %v", records, e)
+ }
}
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go
index 7f4b6dd86..9eb1b0cbb 100644
--- a/internal/provision/reconcile.go
+++ b/internal/provision/reconcile.go
@@ -1226,8 +1226,20 @@ func trashGate(ctx context.Context, s release.Store, what string) func(acct, app
// A record that READS but does not decode — or whose body disagrees with its
// own path — is the other class: a content problem in one object, which is
// skipped and logged (Reconcile's `protect` treatment, minus the registry
-// bookkeeping). One malformed record must not blind a caller to every other.
+// bookkeeping). This lenient variant is unsafe where absence means deletion;
+// those callers need ListAllRecordsComplete.
func ListAllRecords(ctx context.Context, s release.Store) ([]Record, error) {
+ return listAllRecords(ctx, s, false)
+}
+
+// ListAllRecordsComplete refuses malformed or path-disagreeing recognized
+// instance records instead of returning a smaller successful set. It retains
+// the existing key recognition and does not add schema or shape validation.
+func ListAllRecordsComplete(ctx context.Context, s release.Store) ([]Record, error) {
+ return listAllRecords(ctx, s, true)
+}
+
+func listAllRecords(ctx context.Context, s release.Store, complete bool) ([]Record, error) {
known, readFailed, err := knownAccounts(ctx, s)
if err != nil {
return nil, err
@@ -1254,10 +1266,16 @@ func ListAllRecords(ctx context.Context, s release.Store) ([]Record, error) {
}
rec, err := DecodeRecord(raw)
if err != nil {
+ if complete {
+ return nil, fmt.Errorf("provision: read record %s: %w", key, err)
+ }
log.Printf("provision: skipping record %s: malformed: %v", key, err)
continue
}
if rec.Host != host || rec.Account != acct || rec.App != app {
+ if complete {
+ return nil, fmt.Errorf("provision: read record %s: %w", key, ErrMaintenanceProjection)
+ }
log.Printf("provision: skipping record %s: disagrees with its own path (host=%q account=%q app=%q)",
key, rec.Host, rec.Account, rec.App)
continue
diff --git a/internal/steering/completeness_control_test.go b/internal/steering/completeness_control_test.go
new file mode 100644
index 000000000..4c63610bd
--- /dev/null
+++ b/internal/steering/completeness_control_test.go
@@ -0,0 +1,35 @@
+package steering
+
+import (
+ "context"
+ "reflect"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// This demonstrates bounded deletion from an incomplete successful route set;
+// it passes both before and after the upstream completeness fix.
+func TestConvergeDeploymentIncompleteRoutesDeletionControl(t *testing.T) {
+ logs := captureLog(t)
+ f := newFake()
+ edges := [][]byte{edgesBody(t, "dublin", "1.2.3.4", "eu-west-1"), edgesBody(t, "oregon", "5.6.7.8", "us-west-2")}
+ if err := ConvergeDeployment(context.Background(), f, testZone, []registry.Route{steeredRoute("app.example.com"), steeredRoute("other.example.com")}, edges, "example.com"); err != nil {
+ t.Fatal(err)
+ }
+ before := []string{"app.example.com.|A|dublin", "app.example.com.|A|oregon", "edge.example.com.|A|dublin", "edge.example.com.|A|oregon", "other.example.com.|A|dublin", "other.example.com.|A|oregon"}
+ if got := recordNames(f.records); !reflect.DeepEqual(got, before) {
+ t.Fatalf("records = %v, want %v", got, before)
+ }
+ if err := ConvergeDeployment(context.Background(), f, testZone, []registry.Route{steeredRoute("other.example.com")}, edges, "example.com"); err != nil {
+ t.Fatal(err)
+ }
+ want := []string{"edge.example.com.|A|dublin", "edge.example.com.|A|oregon", "other.example.com.|A|dublin", "other.example.com.|A|oregon"}
+ if got := recordNames(f.records); !reflect.DeepEqual(got, want) {
+ t.Fatalf("records = %v, want %v", got, want)
+ }
+ if strings.Contains(logs.String(), "refusing") {
+ t.Fatalf("log = %q", logs.String())
+ }
+}