amadan

carlos / platform Public

zdd-readiness: an unreadable record keeps every fleet in the watch's completeness check, and is never URL-polled

ListRecordsComplete names the hosts whose record could not be read; the instance list counts every fleet toward completeness when any was skipped. The deploy watch keeps an absent host's mode, never moves it to URL watching, and stops polling one that drops off the list (Astra plan 6 R3).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude f9be4718b95b869e2fa6ab35af71fccd2df1b771 parent cd7de72
7 files changed, +182 −13
  • deploywatch.go +34 −8
  • deploywatch_test.go +51 −0
  • internal/console/apiinstances.go +11 −2
  • internal/console/rolloutfleet_test.go +48 −0
  • internal/provision/plan9_b8_census_test.go +1 −1
  • internal/provision/provision.go +13 −2
  • internal/provision/provision_test.go +24 −0
diff --git a/deploywatch.go b/deploywatch.go
index 2ca946652..9b0cf298d 100644
--- a/deploywatch.go
+++ b/deploywatch.go
@@ -216,6 +216,11 @@ type watchedHost struct {
// unread is set while the host is undecided because the console's
// rollout join was incomplete, not because the list was cut short.
unread bool
+ // absent is set while the latest answered list does not carry the host
+ // at all: cut short, or a record the console could not read (Astra plan
+ // 6 R3). An absent host is never polled by URL, even if it was being
+ // polled before; it is cleared the next time the host is listed.
+ absent bool
state rolloutState
// fin is the report the host went final on. A final state survives
// missing observations, and later reports that agree the build is
@@ -227,6 +232,12 @@ type watchedHost struct {
printed bool
}
+// urlPolling is whether this sweep requests the host's URL: it is watched by
+// URL, is not yet live, and the console still lists it.
+func (h *watchedHost) urlPolling() bool {
+ return h.mode == modeURL && !h.urlLive && !h.absent
+}
+
func (h *watchedHost) done() bool {
switch h.mode {
case modeURL:
@@ -302,7 +313,7 @@ func (rw rolloutWatch) run(targets []watchTarget) error {
for {
urlPending := false
for _, h := range hosts {
- if h.mode == modeURL && !h.urlLive {
+ if h.urlPolling() {
urlPending = true
}
}
@@ -314,7 +325,7 @@ func (rw rolloutWatch) run(targets []watchTarget) error {
}
var wg sync.WaitGroup
for _, h := range hosts {
- if h.mode != modeURL || h.urlLive {
+ if !h.urlPolling() {
continue
}
n++
@@ -374,7 +385,7 @@ func (rw rolloutWatch) wait(hosts []*watchedHost, lastList, deadline time.Time)
d := rw.interval
urlPending := false
for _, h := range hosts {
- urlPending = urlPending || (h.mode == modeURL && !h.urlLive)
+ urlPending = urlPending || h.urlPolling()
}
if !urlPending {
d = max(d, time.Until(lastList.Add(rw.listEvery)))
@@ -434,9 +445,11 @@ func (rw rolloutWatch) listFailed(err error) error {
// host back to pending, named by what is installed (pendingDetail): the
// watched build no longer runs there, and Review Focus 4 forbids success
// for a replaced build. A failure stays failed: the build did fail.
-// - A host missing from a truncated list keeps its mode and state (R3):
-// the cap cut it off, nothing about it changed, and switching it to URL
-// polling would wake it if it is asleep.
+// - A host missing from the list keeps its mode and state (R3), whether
+// the cap cut it off or the console could not read its record: nothing
+// says it uses the old path, and switching it to URL polling would wake
+// it if it is asleep. It is also no longer polled if it was (absent),
+// until it is listed again (Astra plan 6 R3).
// - A host with no followed report on a list whose rollout join was
// incomplete keeps its mode and state too: its box's report may be the
// one the console could not read (Astra plan 6 R1 #4). An undecided host
@@ -451,6 +464,7 @@ func (rw rolloutWatch) apply(hosts []*watchedHost, l watchList) {
}
for _, h := range hosts {
ro, ok := listed[h.t.Host]
+ h.absent = !ok
switch {
case follows(ro):
if h.mode != modeRollout {
@@ -471,8 +485,14 @@ func (rw rolloutWatch) apply(hosts []*watchedHost, l watchList) {
case h.state != rolloutFailed && ro.Installed != rw.ver:
h.state, h.fin, h.printed = rolloutPending, nil, false
}
- case !ok && l.truncated:
- h.unread = false
+ case !ok:
+ // Absent from this list: cut short, or a record the console could
+ // not read. Neither says the host uses the old path, so it keeps
+ // its mode and state and is never moved to URL watching here
+ // (Astra plan 6 R3). On an untruncated list the console left it
+ // out because it could not read something, which is what
+ // watchPendingUnread says.
+ h.unread = !l.truncated
case h.mode == modeURL && h.urlLive:
// Live by URL is final too, until a report says otherwise.
case l.incomplete:
@@ -589,6 +609,12 @@ func (rw rolloutWatch) pendingDetail(h *watchedHost) string {
return watchPendingUnread
case h.mode == modeUnknown:
return watchPendingUnlisted
+ case h.mode == modeURL && h.absent && h.urlDiag == "":
+ // Dropped off the list before any poll said anything.
+ if h.unread {
+ return watchPendingUnread
+ }
+ return watchPendingUnlisted
case h.mode == modeURL:
return h.urlDiag
case h.dark:
diff --git a/deploywatch_test.go b/deploywatch_test.go
index 1bd7805ba..1b2ca0ece 100644
--- a/deploywatch_test.go
+++ b/deploywatch_test.go
@@ -877,3 +877,54 @@ func TestWatchRolloutUnknownUnavailableCodeIsNeverPolled(t *testing.T) {
t.Fatalf("%d HTTP polls, err = %v", hits.Load(), err)
}
}
+
+// Astra plan 6 R3: a target missing from a complete, untruncated list is not
+// evidence that it uses the old path (the console may simply not have been
+// able to read its record). It is never polled, and the timeout names it.
+func TestWatchRolloutNeverPollsATargetMissingFromACompleteList(t *testing.T) {
+ var calls atomic.Int64
+ settling := &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}
+ list := scripted(&calls, sweep{recs: []apiInstanceRecord{rec("a.test", settling)}})
+ rw := testWatch(io.Discard, list, noURL(t))
+ onURL := 0
+ rw.onURL = func() { onURL++ }
+ err := rw.run(targetsFor("a.test", "b.test"))
+ if onURL != 0 {
+ t.Fatalf("onURL ran %d times: a missing host was watched by URL", onURL)
+ }
+ if err == nil || !strings.Contains(err.Error(), "\n b.test: "+watchPendingUnread) {
+ t.Fatalf("err = %v, want a timeout naming b.test", err)
+ }
+}
+
+// A host first listed with no report on a complete join is URL-watched. If it
+// then drops off the list (its record became unreadable), the watch must stop
+// requesting it: the request would wake it if it is asleep.
+func TestWatchRolloutStopsPollingAURLHostThatDropsOffTheList(t *testing.T) {
+ var calls, polls, atDrop atomic.Int64
+ settling := &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}
+ list := func(time.Time) (watchList, error) {
+ if calls.Add(1) == 1 {
+ return watchList{recs: []apiInstanceRecord{rec("a.test", settling), rec("b.test", nil)}}, nil
+ }
+ atDrop.CompareAndSwap(0, polls.Load()+1) // +1 so a snapshot of zero polls is not mistaken for unset
+ return watchList{recs: []apiInstanceRecord{rec("a.test", settling)}}, nil
+ }
+ check := func(tg watchTarget, _ int64) (bool, string) {
+ if tg.Host != "b.test" {
+ t.Errorf("%s was polled over HTTP", tg.Host)
+ }
+ polls.Add(1)
+ return false, "not yet"
+ }
+ err := testWatch(io.Discard, list, check).run(targetsFor("a.test", "b.test"))
+ if err == nil {
+ t.Fatal("the watch succeeded; a was never final")
+ }
+ if atDrop.Load() < 2 {
+ t.Fatalf("b was never polled before it dropped off the list (snapshot %d)", atDrop.Load())
+ }
+ if got, want := polls.Load(), atDrop.Load()-1; got != want {
+ t.Fatalf("%d polls, %d when b dropped off the list: b was polled after it was gone", got, want)
+ }
+}
diff --git a/internal/console/apiinstances.go b/internal/console/apiinstances.go
index 294643731..b3184ca63 100644
--- a/internal/console/apiinstances.go
+++ b/internal/console/apiinstances.go
@@ -627,13 +627,22 @@ func (s *Server) handleAPIInstanceList(w http.ResponseWriter, r *http.Request) {
}
ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
defer cancel()
- recs, truncated, err := provision.ListRecords(ctx, s.cfg.Releases, acct.Sqid, app, instanceListMax)
+ recs, truncated, skipped, err := provision.ListRecordsComplete(ctx, s.cfg.Releases, acct.Sqid, app, instanceListMax)
if err != nil {
log.Printf("console: api instances list %s/%s: %v", acct.Sqid, app, err)
jsonError(w, http.StatusBadGateway, "could not list instances")
return
}
- join, joinComplete := s.loadRolloutJoin(ctx, acct.Sqid, time.Now(), recordFleets(recs))
+ relevant := recordFleets(recs)
+ if len(skipped) > 0 {
+ // A skipped record's placement is unknown, so every fleet the account
+ // has, and the platform, counts toward completeness (nil means all).
+ // Astra plan 6 R3: an unreadable record must not drop its fleet.
+ log.Printf("console: api instances list %s/%s: %d record(s) unreadable %v; every fleet counts toward the rollout join",
+ acct.Sqid, app, len(skipped), skipped)
+ relevant = nil
+ }
+ join, joinComplete := s.loadRolloutJoin(ctx, acct.Sqid, time.Now(), relevant)
served := loadEdgesJoin(ctx, s.cfg.Releases)
params := r.URL.Query()
q := parseInstanceQuery(params)
diff --git a/internal/console/rolloutfleet_test.go b/internal/console/rolloutfleet_test.go
index ccc08bc5d..fedc1ddd2 100644
--- a/internal/console/rolloutfleet_test.go
+++ b/internal/console/rolloutfleet_test.go
@@ -190,6 +190,54 @@ func TestAnUnrelatedUnreadableFleetDoesNotMakeTheListIncomplete(t *testing.T) {
}
}
+// Astra plan 6 R3: an unreadable record must never drop its fleet from the
+// completeness check. The record's host sits on fleet/pi, whose only
+// snapshot is unreadable too. The readable record is on the platform, so
+// without the fix no fleet is relevant and the list would claim to be
+// complete, sending the deploy watch to poll URLs it must not poll.
+func TestAnUnreadableRecordKeepsEveryFleetInTheCompletenessCheck(t *testing.T) {
+ fx := newConsoleFixture(t)
+ const piHost = "pi-host.example.test"
+ placedRecord(t, fx, "kass", heldHost, "ie")
+ placedRecord(t, fx, "kass", piHost, registry.PlacementTarget("pi"))
+ reg, err := registry.Open(fx.regPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ if err := reg.UpsertFleet(registry.Fleet{Name: "pi", Account: fx.sqid, Bucket: "pi-bucket"}); err != nil {
+ t.Fatal(err)
+ }
+ box := status.CollectBox(nil, status.Probes{MachineID: "box1"}, time.Now())
+ box.Metrics = &status.BoxMetrics{RouteVersions: map[string]string{heldHost: "v1"}}
+ if err := status.WriteBox(context.Background(), fx.rel, box); err != nil {
+ t.Fatal(err)
+ }
+ token := cliToken(t, fx, "paul@keymail.dev", "code-unreadable-record")
+
+ // Both readable and complete first: the platform record's fleet set is
+ // empty, but the pi record is placed on pi, whose snapshot is fine.
+ if _, incomplete := listRollout(t, fx, token, "kass", heldHost); incomplete {
+ t.Fatal("a healthy app was reported incomplete before anything was broken")
+ }
+
+ // Now the pi record and the pi snapshot both go unreadable.
+ key := provision.RecordKey(release.OpsAccount, "kass", piHost)
+ _, etag, err := fx.rel.Get(context.Background(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.rel.Put(context.Background(), key, []byte("{not json"), etag); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.rel.Put(context.Background(), "fleets/pi/status/boxes/broken.json", []byte("{not json"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, incomplete := listRollout(t, fx, token, "kass", heldHost); !incomplete {
+ t.Fatal("an unreadable record dropped its fleet from the completeness check")
+ }
+}
+
// The fleets that could hold an app's reports: where its records are placed,
// and both ends of a move in flight. A platform region names none.
func TestRecordFleetsNamesPlacementsAndMoveEnds(t *testing.T) {
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index ce639dade..1a3bbf2bb 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -846,7 +846,7 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/provision/provision.go", "DeleteRecord", "key-site", "key:RecordKey", "67e8f77d784d3e84626f2856bd45d219759ba4b603bc3daebbb9e315131c4945"},
{"internal/provision/provision.go", "claimRawUpdateRecord", "record-writer", "key:RecordKey,key:WriteRecord,key:claimRawUpdateRecord", "26adfc4074be815031c8aee929914f60537de925a6d1aba6737772b149b8987c"},
{"internal/provision/provision.go", "GetRecordETag", "key-site", "key:RecordKey", "d9f71e5a52091dce7ea6b6aa3e5d725456d5944efa180727674e7df0006d4fa5"},
- {"internal/provision/provision.go", "ListRecords", "key-site", "key:RecordsPrefix", "d2cc5fb016d58b8091277545187a96c184e178ea5677e72890dae0fef2d68e93"},
+ {"internal/provision/provision.go", "ListRecordsComplete", "key-site", "key:RecordsPrefix", "c04c10f73680b0f4e0a00caf6c5b39f7812ebd72dae89c87f076a8167a16104b"},
{"internal/provision/provision.go", "InstanceDir", "key-site", "lit:instances", "4defc7f996c83afe8d2d2efddd31f4de79d94a5c84e7ee99e8d639ca03d7bf81"},
{"internal/provision/provision.go", "InstanceRunDir", "key-site", "lit:instances", "c0e58b35ea6c773c71524a7d9f5ab22c9a7bec46d0e0ab230784b935f77186c9"},
{"internal/provision/provision.go", "NestedPaths", "key-site", "lit:instances", "64a91261fe43412aab90175f5af2333c35a5fef7232ce20c64fd678a27130826"},
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index 98d688b7d..19445442e 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -2623,13 +2623,23 @@ func GetRecordETag(ctx context.Context, s release.Store, acct, app, host string)
// truncated when any keys exist, the same "cap decided on key count" rule
// applied at its floor.
func ListRecords(ctx context.Context, s release.Store, acct, app string, max int) (recs []Record, truncated bool, err error) {
+ recs, truncated, _, err = ListRecordsComplete(ctx, s, acct, app, max)
+ return recs, truncated, err
+}
+
+// ListRecordsComplete is ListRecords that also names each host whose record
+// could not be read (a store error, a malformed body, a body that disagrees
+// with its path). A caller that must know whether the list is the whole app
+// (the deploy watch's completeness check, Astra plan 6 R3) reads skipped.
+// Hosts cut by the cap are not skipped; truncated reports those.
+func ListRecordsComplete(ctx context.Context, s release.Store, acct, app string, max int) (recs []Record, truncated bool, skipped []string, err error) {
if max < 0 {
max = 0
}
prefix := RecordsPrefix(acct, app)
keys, err := s.List(ctx, prefix)
if err != nil {
- return nil, false, err
+ return nil, false, nil, err
}
var hosts []string
for _, key := range keys {
@@ -2651,11 +2661,12 @@ func ListRecords(ctx context.Context, s release.Store, acct, app string, max int
for _, host := range hosts {
rec, err := GetRecord(ctx, s, acct, app, host)
if err != nil {
+ skipped = append(skipped, host)
continue
}
recs = append(recs, *rec)
}
- return recs, truncated, nil
+ return recs, truncated, skipped, nil
}
// SocketDir is where a hibernating instance's control socket lives — the
diff --git a/internal/provision/provision_test.go b/internal/provision/provision_test.go
index e7ba07d95..d233c2b9b 100644
--- a/internal/provision/provision_test.go
+++ b/internal/provision/provision_test.go
@@ -404,6 +404,30 @@ func TestManifestRoundTrip(t *testing.T) {
}
}
+// Astra plan 6 R3: the deploy watch must know when the list it was handed is
+// not the whole app, so a record that cannot be read is named, not dropped
+// silently.
+func TestListRecordsCompleteNamesTheSkippedHost(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ rec := Record{Schema: 1, Host: "a.example.com", Slug: "a", App: "app", Account: "acct", Channel: "stable", CreatedAt: time.Now().UTC()}
+ if err := CreateRecord(ctx, s, rec); err != nil {
+ t.Fatalf("CreateRecord: %v", err)
+ }
+ if _, err := s.Put(ctx, RecordKey("acct", "app", "b.example.com"), []byte("not json"), ""); err != nil {
+ t.Fatal(err)
+ }
+ recs, truncated, skipped, err := ListRecordsComplete(ctx, s, "acct", "app", 10)
+ if err != nil || truncated || len(recs) != 1 || len(skipped) != 1 || skipped[0] != "b.example.com" {
+ t.Fatalf("recs %d truncated %v skipped %v err %v", len(recs), truncated, skipped, err)
+ }
+ // ListRecords is the same list without the skipped names.
+ recs, truncated, err = ListRecords(ctx, s, "acct", "app", 10)
+ if err != nil || truncated || len(recs) != 1 {
+ t.Fatalf("ListRecords: recs %d truncated %v err %v", len(recs), truncated, err)
+ }
+}
+
func TestListRecords(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())