amadan
console,provision: instance create carries route shape; PATCH updates addrs/probe; owned rows converge upstreams
The console's instance API grows the three surfaces the multi-upstream
design needs, and the reconcile pass grows the converge that makes them
mean anything on a box.
- POST .../instances gains kind, addrs and probe (all optional, all
defaulting to the exact route this door minted before they existed) and
runs provision.ValidateShape at the door, so the caller gets the same
sentence the box's pass would print. Policy is unmoved: pool and channel
still come from the manifest, and the host suffix rule is unchanged — a
shape field is never a way to claim a hostname the manifest refuses.
- NEW PATCH /api/cli/apps/{account}/{app}/instances/{host} repoints a live
record's upstream set and probe. Absent/empty/null are three different
requests; a body asking for neither field is a 400, not a no-op write.
The write-back is guarded by the etag the record was read under
(provision.UpdateRecord, new) so a lost race is a 409 rather than a
clobber of bytes nobody read, and the serial is bumped once, strictly
after the write lands. Every refusal before the principal is proven —
including a host that names no record — is the family's byte-identical
404.
- GET .../instances joins per-upstream probe state from the box snapshots
(status.BoxMetrics.Upstreams) beside staged_version, under the same
BoxStale rule and through one loader so the two joins cannot disagree.
Omitempty: every single-upstream record parses exactly as before.
- provision.convergeOwned now converges Addrs/Health on owned rows through
registry.SetUpstreams — until now SetUpstreams had no non-test caller and
the owned-row drift log never looked at those columns, so a record whose
upstreams changed after mint kept the old set forever, silently. Mint and
converge share one derivation (upstreamCols) so the two cannot drift.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
7 files changed,
+921
−31
internal/console/apiinstances.go+191 −12internal/console/apiinstances_test.go+436 −0internal/console/authcli.go+5 −0internal/provision/provision.go+52 −5internal/provision/provision_test.go+53 −0internal/provision/reconcile.go+57 −14internal/provision/reconcile_test.go+127 −0
diff --git a/internal/console/apiinstances.go b/internal/console/apiinstances.go| index 52e72dc27..aaa6a1ca3 100644 |
| --- a/internal/console/apiinstances.go |
| +++ b/internal/console/apiinstances.go |
| @@ -21,6 +21,7 @@ package console |
| import ( |
| "context" |
| + "encoding/json" |
| "errors" |
| "log" |
| "net/http" |
| @@ -169,12 +170,23 @@ type instanceListRecord struct { |
| // status/boxes/<label>.json (internal/status's bucket-private Metrics |
| // block — see its package doc). Empty both for a host no box has ever |
| // reported AND for one whose last-reporting box has gone stale (see |
| - // stagedVersions below, which asks status.BoxStale the same question |
| + // loadBoxJoin below, which asks status.BoxStale the same question |
| // the operator fleet page does): fresh-ish for as long as some box keeps |
| // reporting it, but a box that has gone silent drops OUT of the join |
| // rather than going on serving that box's last-known version as if it |
| // were current. The deploy watch does NOT read this field. |
| StagedVersion string `json:"staged_version"` |
| + // Upstreams is the per-upstream probe state the box's edge currently |
| + // believes for this host (status.BoxMetrics.Upstreams), joined from the |
| + // same snapshots and under the same staleness rule as StagedVersion — a |
| + // box that has gone silent drops out rather than going on answering for |
| + // upstreams it may no longer be probing. It is what lets `carlos |
| + // instances` say WHICH upstream is down without a box session, and it is |
| + // omitempty because that is the honest shape for the overwhelming |
| + // majority of records: only a route with more than one addr (or an |
| + // explicit probe block) is probed at all, so every single-upstream record |
| + // in production carries no such key and parses exactly as it always did. |
| + Upstreams []status.UpstreamState `json:"upstreams,omitempty"` |
| // Health is the same one-word state the Running tab's column shows, |
| // computed by instanceHealth below from the registry row joined with the |
| // metrics sweep. Additive on the wire (a client that never heard of it |
| @@ -283,8 +295,22 @@ func (s *Server) instanceHealth(ctx context.Context, recs []provision.Record) ma |
| return out |
| } |
| -// stagedVersions builds a host -> version map from every NON-STALE box's own |
| -// private RouteVersions (status/boxes/*.json), through status.LoadBoxes — |
| +// boxJoin is everything the list annotates a record with from the FLEET's own |
| +// snapshots, gathered in one pass: the staged version and the per-upstream |
| +// probe table, both host-keyed. |
| +// |
| +// One struct and one loader rather than two functions, because the two joins |
| +// must vote on staleness with the same reading of the same objects: a second |
| +// LoadBoxes call could see a box go stale between them and answer a record's |
| +// version from a box whose upstream table it had just refused, which is a |
| +// disagreement no reader could diagnose. |
| +type boxJoin struct { |
| + versions map[string]string |
| + upstreams map[string][]status.UpstreamState |
| +} |
| + |
| +// loadBoxJoin builds boxJoin from every NON-STALE box's own |
| +// private Metrics block (status/boxes/*.json), through status.LoadBoxes — |
| // the same loader the operator fleet page's HostRows reads through |
| // (internal/console/view.go) — and status.BoxStale, the fleet's one |
| // staleness rule (internal/status/fleet.go: "Every caller ... asks this |
| @@ -307,25 +333,28 @@ func (s *Server) instanceHealth(ctx context.Context, recs []provision.Record) ma |
| // today — instanceGate already refuses those with a 503 before this ever |
| // runs — but the nil check is kept here too so this helper degrades safely |
| // for any future caller that reaches it before that gate. |
| -func stagedVersions(ctx context.Context, s release.Store, now time.Time) map[string]string { |
| +func loadBoxJoin(ctx context.Context, s release.Store, now time.Time) boxJoin { |
| if s == nil { |
| - return nil |
| + return boxJoin{} |
| } |
| boxes, err := status.LoadBoxes(ctx, s) |
| if err != nil { |
| - log.Printf("console: api instances: could not load status boxes for staged_version: %v", err) |
| - return nil |
| + log.Printf("console: api instances: could not load status boxes for the list join: %v", err) |
| + return boxJoin{} |
| } |
| - versions := map[string]string{} |
| + j := boxJoin{versions: map[string]string{}, upstreams: map[string][]status.UpstreamState{}} |
| for _, b := range boxes { |
| if b.Metrics == nil || status.BoxStale(b, now, status.SnapshotPeriod) { |
| continue |
| } |
| for host, v := range b.Metrics.RouteVersions { |
| - versions[host] = v |
| + j.versions[host] = v |
| + } |
| + for host, ups := range b.Metrics.Upstreams { |
| + j.upstreams[host] = ups |
| } |
| } |
| - return versions |
| + return j |
| } |
| // handleAPIInstanceList answers `carlos instances list`: this app's records, |
| @@ -354,7 +383,7 @@ func (s *Server) handleAPIInstanceList(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusBadGateway, "could not list instances") |
| return |
| } |
| - versions := stagedVersions(ctx, s.cfg.Releases, time.Now()) |
| + join := loadBoxJoin(ctx, s.cfg.Releases, time.Now()) |
| params := r.URL.Query() |
| q := parseInstanceQuery(params) |
| // The default stays the whole cap, so a client that sends no limit gets |
| @@ -379,7 +408,8 @@ func (s *Server) handleAPIInstanceList(w http.ResponseWriter, r *http.Request) { |
| if q.Health != "" && h != q.Health { |
| continue |
| } |
| - matched = append(matched, instanceListRecord{Record: rec, StagedVersion: versions[rec.Host], Health: h}) |
| + matched = append(matched, instanceListRecord{Record: rec, |
| + StagedVersion: join.versions[rec.Host], Health: h, Upstreams: join.upstreams[rec.Host]}) |
| } |
| total := len(matched) |
| // provision.ListRecords returns records host-sorted and nothing above |
| @@ -461,6 +491,20 @@ func (s *Server) handleAPIInstanceCreate(w http.ResponseWriter, r *http.Request) |
| Host string `json:"host"` |
| Region string `json:"region"` |
| Channel string `json:"channel"` |
| + // The route SHAPE fields (route-records design §3, HA design §II). |
| + // All optional, all defaulting to the exact route this door minted |
| + // before they existed, and all gated by provision.ValidateShape |
| + // below — the same door the reconcile pass re-checks them at, because |
| + // neither side trusts the other about what is in the bucket. |
| + // |
| + // What is deliberately NOT here: pool (manifest policy), hibernate, |
| + // unit and environment (operator columns with their own verbs), and |
| + // anything that could widen which HOST this app may claim — the |
| + // suffix check below is unchanged, so a "kind":"static" body cannot |
| + // reach for a hostname outside the app's own domain. |
| + Kind string `json:"kind"` |
| + Addrs []string `json:"addrs"` |
| + Health *provision.Health `json:"probe"` |
| } |
| if !readJSON(w, r, &req) { |
| return |
| @@ -558,6 +602,20 @@ func (s *Server) handleAPIInstanceCreate(w http.ResponseWriter, r *http.Request) |
| // records are read on boxes in other regions. |
| CreatedAt: time.Now().UTC(), |
| CreatedBy: principal, |
| + Kind: strings.TrimSpace(req.Kind), |
| + Addrs: req.Addrs, |
| + Health: req.Health, |
| + } |
| + // The shape rails, at the door — kind vocabulary, one upstream shape per |
| + // record, addr syntax, and the combinations that would wedge a host (a |
| + // static route with an upstream, a hibernating addr-backed one). The |
| + // message is ValidateShape's own: it names WHICH upstream is wrong, which |
| + // is the only thing the caller can act on, and it is the identical |
| + // sentence the box's reconcile pass would print if a record carrying this |
| + // shape ever reached it from another writer. |
| + if err := provision.ValidateShape(rec); err != nil { |
| + jsonError(w, http.StatusBadRequest, err.Error()) |
| + return |
| } |
| if err := provision.CreateRecord(ctx, s.cfg.Releases, rec); err != nil { |
| if errors.Is(err, release.ErrCASConflict) { |
| @@ -584,6 +642,127 @@ func (s *Server) handleAPIInstanceCreate(w http.ResponseWriter, r *http.Request) |
| writeJSON(w, http.StatusCreated, rec) |
| } |
| +// handleAPIInstanceUpdate repoints a live instance's UPSTREAM SET and probe |
| +// config — the two fields of a record that are meant to change while the |
| +// instance keeps serving (HA design, issue #152). Everything else about a |
| +// record stays where it is: identity (host/app/account) is checked at create |
| +// and never touched again, policy (pool, domain) is the manifest's, and |
| +// region has its own writer in placement. |
| +// |
| +// It is a PATCH, and the body's two fields are independently optional, because |
| +// "leave the probe alone while I add an upstream" is the ordinary request. A |
| +// body carrying NEITHER is refused rather than treated as a no-op write: it |
| +// would cost a bucket round trip and a fleet-wide serial bump to change |
| +// nothing. |
| +// |
| +// The write is guarded by the etag the record was READ under |
| +// (provision.UpdateRecord): the caller decided this upstream set given the |
| +// record they were shown, so a concurrent writer means the decision was made |
| +// about bytes that no longer exist, and the honest answer is 409 — the same |
| +// answer the enable door gives for the same reason. |
| +// |
| +// Every refusal before the principal is proven is the family's byte-identical |
| +// 404 (instanceGate), including a host that names no record: a record's |
| +// existence is not a thing to probe for here any more than it is on GET. |
| +func (s *Server) handleAPIInstanceUpdate(w http.ResponseWriter, r *http.Request) { |
| + principal, acct, app, ok := s.instanceGate(w, r) |
| + if !ok { |
| + return |
| + } |
| + host := r.PathValue("host") |
| + // Shape-checked before the key builder, exactly as GET does it and for |
| + // the same reason: a {host} segment arrives percent-decoded, so |
| + // "..%2f..%2fchannels" reaches this handler as a path with separators in |
| + // it and RecordKey would happily join it. The manifest's suffix rule is |
| + // deliberately NOT applied — that is create-time policy ("which hosts may |
| + // this app claim"), and an app that has since moved domains must still be |
| + // able to repoint the upstreams of the instances it already has. |
| + if !provision.ValidHostShape(host) { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + // Addrs is a POINTER and Probe a RawMessage because absent, empty and |
| + // null are three different requests (the apiConfigApply precedent): an |
| + // omitted field leaves what the record carries, `"addrs":[]` clears the |
| + // upstream set, and `"probe":null` clears the probe while `"probe":{}` |
| + // asks for the platform defaults. |
| + var req struct { |
| + Addrs *[]string `json:"addrs"` |
| + Probe json.RawMessage `json:"probe"` |
| + } |
| + if !readJSON(w, r, &req) { |
| + return |
| + } |
| + if req.Addrs == nil && req.Probe == nil { |
| + jsonError(w, http.StatusBadRequest, |
| + "nothing to update — send addrs, probe, or both") |
| + return |
| + } |
| + var probe *provision.Health |
| + if req.Probe != nil { |
| + if err := json.Unmarshal(req.Probe, &probe); err != nil { |
| + jsonError(w, http.StatusBadRequest, |
| + `invalid probe — want an object like {"path":"/healthz","status":200} or null to stop probing`) |
| + return |
| + } |
| + } |
| + ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout) |
| + defer cancel() |
| + rec, etag, err := provision.GetRecordETag(ctx, s.cfg.Releases, acct.Sqid, app, host) |
| + if errors.Is(err, os.ErrNotExist) { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + if err != nil { |
| + log.Printf("console: api instances update read %s/%s/%s: %v", acct.Sqid, app, host, err) |
| + jsonError(w, http.StatusBadGateway, "could not read the instance") |
| + return |
| + } |
| + if req.Addrs != nil { |
| + // This door states the upstream set WHOLE, so it also retires the |
| + // legacy single-addr spelling of the same fact. Leaving Addr standing |
| + // would make every patch of a record minted before Addrs existed fail |
| + // provision.ValidateShape's "addr and addrs both set" rail — a refusal |
| + // about a field the caller never sent, on the exact records (kass, |
| + // keymail, the consoles) this feature exists to give a second upstream |
| + // to. Clearing it is not a widening: AddrsOrDefault already folds Addr |
| + // into the same one-element set this replaces. |
| + rec.Addrs, rec.Addr = *req.Addrs, "" |
| + } |
| + if req.Probe != nil { |
| + rec.Health = probe |
| + } |
| + // The same door the create path and the box's reconcile pass check |
| + // against, so a shape that could not have been created cannot be patched |
| + // into existence either — and the caller gets ValidateShape's own |
| + // sentence, which names the offending upstream. |
| + if err := provision.ValidateShape(*rec); err != nil { |
| + jsonError(w, http.StatusBadRequest, err.Error()) |
| + return |
| + } |
| + // provision.UpdateRecord marshals the whole record through the canonical |
| + // writer: nothing here hand-builds an object body, so upstream ORDER — |
| + // which the reconcile pass's shapeMismatch compares verbatim against the |
| + // registry's joined column — lands exactly as the caller stated it. |
| + if err := provision.UpdateRecord(ctx, s.cfg.Releases, *rec, etag); err != nil { |
| + if errors.Is(err, release.ErrCASConflict) { |
| + jsonError(w, http.StatusConflict, |
| + "instance "+host+" changed while this request was in flight — read it again and re-apply") |
| + return |
| + } |
| + log.Printf("console: api instances update %s/%s/%s: %v", acct.Sqid, app, host, err) |
| + jsonError(w, http.StatusBadGateway, "could not update the instance — try again") |
| + return |
| + } |
| + s.audit(acct, app, principal, "instance-update", "updated instance "+host+" upstreams (cli)") |
| + // Strictly after the write landed, and exactly once: the record is what a |
| + // box's reconcile leg converges upstreams from (provision.convergeOwned), |
| + // so this announcement is what makes the new set live in seconds instead |
| + // of on the hour. |
| + s.bumpSerial(ctx) |
| + writeJSON(w, http.StatusOK, rec) |
| +} |
| + |
| // handleAPIInstanceDelete retires an instance's route authorization. The |
| // record is deleted; the parked database and blobs are deliberately |
| // untouched (spec §3.2) — retiring a route is not destroying a customer's |
diff --git a/internal/console/apiinstances_test.go b/internal/console/apiinstances_test.go| index fd2454918..d8161768d 100644 |
| --- a/internal/console/apiinstances_test.go |
| +++ b/internal/console/apiinstances_test.go |
| @@ -888,6 +888,442 @@ func TestAPIInstanceListHealthAndFilters(t *testing.T) { |
| } |
| } |
| +// TestAPIInstanceCreateCarriesRouteShape (multi-upstream design, issue |
| +// #152): the create door now carries the record's ROUTE SHAPE — kind, the |
| +// upstream set and the probe config — so an app fronting more than one |
| +// process can be provisioned through the same one write every other instance |
| +// is. Policy is unchanged and stays server-side: pool and channel still come |
| +// from the manifest, and the host still has to be one label under the app's |
| +// own domain (a shape field must never become a way to claim a hostname the |
| +// suffix rule refuses). |
| +func TestAPIInstanceCreateCarriesRouteShape(t *testing.T) { |
| + fx, token := instanceFixture(t, "code-inst-shape") |
| + enableInstances(t, fx, "titogo", titoManifest()) |
| + u := instancesURL(fx, "bab", "titogo") |
| + |
| + status, _, out := rawReq(t, http.MethodPost, u, token, |
| + `{"host":"ha.go.tito.io","region":"ie","kind":"service",`+ |
| + `"addrs":["127.0.0.1:7001","127.0.0.1:7002"],"probe":{"path":"/healthz"}}`) |
| + if status != http.StatusCreated { |
| + t.Fatalf("create with a route shape = %d %s, want 201", status, out) |
| + } |
| + var rec provision.Record |
| + if err := json.Unmarshal([]byte(out), &rec); err != nil { |
| + t.Fatalf("create body %s: %v", out, err) |
| + } |
| + if rec.Kind != "service" || !slices.Equal(rec.Addrs, []string{"127.0.0.1:7001", "127.0.0.1:7002"}) { |
| + t.Fatalf("echoed record = %+v, want the requested kind and upstreams in order", rec) |
| + } |
| + if rec.Health == nil || rec.Health.Path != "/healthz" { |
| + t.Fatalf("echoed probe = %+v, want the requested path", rec.Health) |
| + } |
| + // Manifest policy still wins over anything the body could say. |
| + if rec.Pool != "tito" || rec.Channel != "stable" { |
| + t.Fatalf("record = %+v, want pool and channel from the manifest", rec) |
| + } |
| + |
| + // The bucket copy is what the reconcile pass mints from, so it — not just |
| + // the echo — has to carry the shape, under the record's own wire keys. |
| + stored, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if stored.Kind != "service" || !slices.Equal(stored.Addrs, rec.Addrs) || |
| + stored.Health == nil || stored.Health.Path != "/healthz" { |
| + t.Fatalf("stored record = %+v", stored) |
| + } |
| + raw, _, err := fx.rel.Get(context.Background(), provision.RecordKey("bab", "titogo", "ha.go.tito.io")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !strings.Contains(string(raw), `"probe":`) { |
| + t.Fatalf("stored record does not use the probe wire key: %s", raw) |
| + } |
| + |
| + // A record shape ValidateShape refuses never reaches the bucket, and the |
| + // refusal is ValidateShape's own sentence — the caller has to know WHICH |
| + // upstream was wrong. |
| + for _, c := range []struct{ what, body, want string }{ |
| + {"bad upstream", `{"host":"bad.go.tito.io","region":"ie","kind":"service",` + |
| + `"addrs":["127.0.0.1:7001","not-an-addr"]}`, "host:port"}, |
| + {"duplicate upstream", `{"host":"dup.go.tito.io","region":"ie","kind":"service",` + |
| + `"addrs":["127.0.0.1:7001","127.0.0.1:7001"]}`, "duplicate upstream"}, |
| + {"unknown kind", `{"host":"weird.go.tito.io","region":"ie","kind":"daemon"}`, "instance, service or static"}, |
| + {"service with no upstream", `{"host":"svc.go.tito.io","region":"ie","kind":"service"}`, "must name the upstream"}, |
| + {"static with upstreams", `{"host":"site.go.tito.io","region":"ie","kind":"static",` + |
| + `"addrs":["127.0.0.1:7001"]}`, "proxies to nothing"}, |
| + } { |
| + status, _, out := rawReq(t, http.MethodPost, u, token, c.body) |
| + if status != http.StatusBadRequest { |
| + t.Errorf("%s = %d %s, want 400", c.what, status, out) |
| + continue |
| + } |
| + if !strings.Contains(out, c.want) { |
| + t.Errorf("%s refusal = %s, want it to name %q", c.what, out, c.want) |
| + } |
| + } |
| + keys, err := fx.rel.List(context.Background(), provision.RecordsPrefix("bab", "titogo")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(keys) != 1 { |
| + t.Fatalf("refused creates left records in the bucket: %v", keys) |
| + } |
| + |
| + // A record carrying none of the new fields is byte-identical to what the |
| + // door wrote before they existed — the additivity rule, at the door. |
| + if status, _, out := rawReq(t, http.MethodPost, u, token, |
| + `{"host":"plain.go.tito.io","region":"ie"}`); status != http.StatusCreated { |
| + t.Fatalf("plain create = %d %s, want 201", status, out) |
| + } |
| + plain, _, err := fx.rel.Get(context.Background(), provision.RecordKey("bab", "titogo", "plain.go.tito.io")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, k := range []string{`"kind"`, `"addrs"`, `"probe"`} { |
| + if strings.Contains(string(plain), k) { |
| + t.Errorf("a shapeless create wrote %s into the record: %s", k, plain) |
| + } |
| + } |
| +} |
| + |
| +// updateURL is the PATCH door: one instance's mutable route shape. |
| +func updateURL(fx *consoleFixture, acct, app, host string) string { |
| + return instancesURL(fx, acct, app) + "/" + host |
| +} |
| + |
| +// TestAPIInstanceUpdateUpstreams: the PATCH door rewrites a live record's |
| +// upstream set and probe config in place — the whole point of the |
| +// multi-upstream feature being operable from the CLI — and announces the |
| +// write so boxes converge in seconds. The record is rewritten through the |
| +// canonical marshaler (so upstream ORDER, which shapeMismatch compares |
| +// verbatim, lands stable), everything else about the record is left exactly |
| +// as it was, and a body that asks for nothing is refused rather than costing |
| +// a bucket write and a fleet-wide wake. |
| +func TestAPIInstanceUpdateUpstreams(t *testing.T) { |
| + fx, token := instanceFixture(t, "code-inst-patch") |
| + enableInstances(t, fx, "titogo", titoManifest()) |
| + u := instancesURL(fx, "bab", "titogo") |
| + if status, _, out := rawReq(t, http.MethodPost, u, token, |
| + `{"host":"ha.go.tito.io","region":"ie","kind":"service","addrs":["127.0.0.1:7001"]}`); status != http.StatusCreated { |
| + t.Fatalf("create = %d %s", status, out) |
| + } |
| + created, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + afterCreate := serial(t, fx) |
| + if afterCreate == 0 { |
| + t.Fatalf("create did not bump the serial") |
| + } |
| + |
| + pu := updateURL(fx, "bab", "titogo", "ha.go.tito.io") |
| + status, _, out := rawReq(t, http.MethodPatch, pu, token, |
| + `{"addrs":["127.0.0.1:7002","127.0.0.1:7001"],"probe":{"path":"/healthz","fail_after":5}}`) |
| + if status != http.StatusOK { |
| + t.Fatalf("patch = %d %s, want 200", status, out) |
| + } |
| + var echoed provision.Record |
| + if err := json.Unmarshal([]byte(out), &echoed); err != nil { |
| + t.Fatalf("patch body %s: %v", out, err) |
| + } |
| + if !slices.Equal(echoed.Addrs, []string{"127.0.0.1:7002", "127.0.0.1:7001"}) { |
| + t.Fatalf("echoed addrs = %v, want the requested pair IN THE REQUESTED ORDER", echoed.Addrs) |
| + } |
| + |
| + stored, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Equal(stored.Addrs, []string{"127.0.0.1:7002", "127.0.0.1:7001"}) { |
| + t.Fatalf("stored addrs = %v, want the requested pair in order", stored.Addrs) |
| + } |
| + if stored.Health == nil || stored.Health.Path != "/healthz" || stored.Health.FailAfter != 5 { |
| + t.Fatalf("stored probe = %+v", stored.Health) |
| + } |
| + // Identity, provenance and everything policy owns are untouched: this |
| + // door speaks for two fields and no others. |
| + if stored.Host != created.Host || stored.App != created.App || stored.Account != created.Account || |
| + stored.Slug != created.Slug || stored.Kind != created.Kind || stored.Pool != created.Pool || |
| + stored.Region != created.Region || stored.Channel != created.Channel || |
| + stored.CreatedBy != created.CreatedBy || !stored.CreatedAt.Equal(created.CreatedAt) { |
| + t.Fatalf("patch moved a field it does not own:\n got %+v\nwant %+v", stored, created) |
| + } |
| + // The record is what a box's reconcile leg reads, so the write is |
| + // announced exactly as create's is — once. |
| + if got := serial(t, fx); got != afterCreate+1 { |
| + t.Fatalf("serial after patch = %d, want exactly one bump past %d", got, afterCreate) |
| + } |
| + |
| + // addrs alone leaves the probe standing; probe alone leaves the addrs. |
| + if status, _, out := rawReq(t, http.MethodPatch, pu, token, `{"addrs":["127.0.0.1:7003"]}`); status != http.StatusOK { |
| + t.Fatalf("patch addrs only = %d %s, want 200", status, out) |
| + } |
| + stored, err = provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Equal(stored.Addrs, []string{"127.0.0.1:7003"}) || stored.Health == nil || stored.Health.Path != "/healthz" { |
| + t.Fatalf("addrs-only patch = %+v, want the probe left standing", stored) |
| + } |
| + if status, _, out := rawReq(t, http.MethodPatch, pu, token, `{"probe":{"path":"/up"}}`); status != http.StatusOK { |
| + t.Fatalf("patch probe only = %d %s, want 200", status, out) |
| + } |
| + stored, err = provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Equal(stored.Addrs, []string{"127.0.0.1:7003"}) || stored.Health.Path != "/up" { |
| + t.Fatalf("probe-only patch = %+v, want the addrs left standing", stored) |
| + } |
| + |
| + // Explicit null clears the probe — absent and null are different requests |
| + // on the wire (the apiConfigApply precedent). |
| + if status, _, out := rawReq(t, http.MethodPatch, pu, token, `{"probe":null}`); status != http.StatusOK { |
| + t.Fatalf("patch clearing the probe = %d %s, want 200", status, out) |
| + } |
| + stored, err = provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if stored.Health != nil { |
| + t.Fatalf("probe:null did not clear the probe: %+v", stored.Health) |
| + } |
| + |
| + before := serial(t, fx) |
| + for _, c := range []struct{ what, body, want string }{ |
| + {"neither field", `{}`, "addrs"}, |
| + {"bad upstream", `{"addrs":["nope"]}`, "host:port"}, |
| + {"duplicate upstream", `{"addrs":["127.0.0.1:7003","127.0.0.1:7003"]}`, "duplicate upstream"}, |
| + } { |
| + status, _, out := rawReq(t, http.MethodPatch, pu, token, c.body) |
| + if status != http.StatusBadRequest { |
| + t.Errorf("patch with %s = %d %s, want 400", c.what, status, out) |
| + continue |
| + } |
| + if !strings.Contains(out, c.want) { |
| + t.Errorf("patch with %s refused with %s, want it to name %q", c.what, out, c.want) |
| + } |
| + } |
| + // A refused patch wrote nothing and announced nothing. |
| + unchanged, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Equal(unchanged.Addrs, []string{"127.0.0.1:7003"}) { |
| + t.Errorf("a refused patch changed the record: %+v", unchanged) |
| + } |
| + if got := serial(t, fx); got != before { |
| + t.Errorf("serial after refused patches = %d, want it unmoved from %d", got, before) |
| + } |
| + |
| + // A host with no record is the family's byte-identical 404 — a record's |
| + // existence is not a thing to probe for through this door either — and so |
| + // is a host that could never name one. |
| + want := canonicalNotFound(t) |
| + status, h, out := rawReq(t, http.MethodPatch, updateURL(fx, "bab", "titogo", "missing.go.tito.io"), |
| + token, `{"addrs":["127.0.0.1:7001"]}`) |
| + assertNotFound(t, "patch of a host with no record", want, status, h, out) |
| + status, h, out = rawReq(t, http.MethodPatch, instancesURL(fx, "bab", "titogo")+"/..%2f..%2fchannels", |
| + token, `{"addrs":["127.0.0.1:7001"]}`) |
| + assertNotFound(t, "patch of a traversal host", want, status, h, out) |
| +} |
| + |
| +// TestAPIInstanceUpdateRefusalMatrix: the NEW verb joins the family's refusal |
| +// discipline — every way of not being allowed to touch this record answers |
| +// with the bytes http.NotFound writes, so a probe cannot learn that an |
| +// account, an app or an instance exists. This is the gate rule this repo has |
| +// paid for; a new route is exactly where it gets forgotten. |
| +func TestAPIInstanceUpdateRefusalMatrix(t *testing.T) { |
| + fx, member := instanceFixture(t, "code-inst-patch-refuse") |
| + enableInstances(t, fx, "titogo", titoManifest()) |
| + if status, _, out := rawReq(t, http.MethodPost, instancesURL(fx, "bab", "titogo"), member, |
| + `{"host":"ha.go.tito.io","region":"ie","kind":"service","addrs":["127.0.0.1:7001"]}`); status != http.StatusCreated { |
| + t.Fatalf("create = %d %s", status, out) |
| + } |
| + now := time.Now() |
| + if err := fx.st.CreateAccount(42, "cdef", "widgets", "bob@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := fx.st.CreateAppIn(42, "titogo", "bob@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + appTok, err := fx.st.NewAppToken(OpsAccountID, "titogo", "alice@keymail.dev", "signup", now) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + eve := cliToken(t, fx, "eve@keymail.dev", "code-inst-patch-eve") |
| + |
| + body := `{"addrs":["10.0.0.1:9999"]}` |
| + want := canonicalNotFound(t) |
| + for _, c := range []struct{ what, url, token string }{ |
| + {"non-member", updateURL(fx, "bab", "titogo", "ha.go.tito.io"), eve}, |
| + {"app token, foreign account", updateURL(fx, "cdef", "titogo", "ha.go.tito.io"), appTok}, |
| + {"app token, foreign app", updateURL(fx, "bab", "other", "ha.go.tito.io"), appTok}, |
| + {"app token, own account by name", updateURL(fx, "ops", "titogo", "ha.go.tito.io"), appTok}, |
| + {"unknown account", updateURL(fx, "zzzz", "titogo", "ha.go.tito.io"), member}, |
| + {"unclaimed app", updateURL(fx, "bab", "ghost-app", "ha.go.tito.io"), member}, |
| + } { |
| + status, h, out := rawReq(t, http.MethodPatch, c.url, c.token, body) |
| + assertNotFound(t, "patch as "+c.what, want, status, h, out) |
| + } |
| + // No credential at all is the 401 every CLI door gives. |
| + if status, _, out := rawReq(t, http.MethodPatch, updateURL(fx, "bab", "titogo", "ha.go.tito.io"), "", body); status != http.StatusUnauthorized { |
| + t.Errorf("patch with no token = %d %s, want 401", status, out) |
| + } |
| + // Nothing above touched the record. |
| + rec, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Equal(rec.Addrs, []string{"127.0.0.1:7001"}) { |
| + t.Fatalf("a refused patch reached the record: %+v", rec) |
| + } |
| + // An app token MAY patch its own app's instance: this is the door a |
| + // signup-time app uses to repoint its own upstreams. |
| + if status, _, out := rawReq(t, http.MethodPatch, updateURL(fx, "bab", "titogo", "ha.go.tito.io"), |
| + appTok, body); status != http.StatusOK { |
| + t.Fatalf("app token patching its own app = %d %s, want 200", status, out) |
| + } |
| + // And a console with no deployment bucket answers 503, not a panic. |
| + fx.srv.cfg.Releases = nil |
| + if status, _, out := rawReq(t, http.MethodPatch, updateURL(fx, "bab", "titogo", "ha.go.tito.io"), |
| + member, body); status != http.StatusServiceUnavailable { |
| + t.Errorf("patch with no bucket = %d %s, want 503", status, out) |
| + } |
| +} |
| + |
| +// staleETagStore hands every reader an etag no stored object carries — what a |
| +// concurrent writer looks like from inside one request: the record was read, |
| +// and by the time the write-back went out the bytes had moved on. |
| +type staleETagStore struct{ release.Store } |
| + |
| +func (s staleETagStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + body, _, err := s.Store.Get(ctx, key) |
| + return body, "moved-on", err |
| +} |
| + |
| +// TestAPIInstanceUpdateLostRace: the guarded write-back surfaces a lost race |
| +// as a 409 rather than overwriting whatever landed in between. The read and |
| +// the write are one decision — a member sent the upstream set they wanted |
| +// GIVEN the record they saw — so silently re-applying it to a body nobody |
| +// read would be exactly the clobber the etag exists to prevent. Nothing is |
| +// announced: no write landed, so there is no news. |
| +func TestAPIInstanceUpdateLostRace(t *testing.T) { |
| + fx, token := instanceFixture(t, "code-inst-patch-race") |
| + enableInstances(t, fx, "titogo", titoManifest()) |
| + if status, _, out := rawReq(t, http.MethodPost, instancesURL(fx, "bab", "titogo"), token, |
| + `{"host":"ha.go.tito.io","region":"ie","kind":"service","addrs":["127.0.0.1:7001"]}`); status != http.StatusCreated { |
| + t.Fatalf("create = %d %s", status, out) |
| + } |
| + before := serial(t, fx) |
| + |
| + fx.srv.cfg.Releases = staleETagStore{fx.rel} |
| + status, _, out := rawReq(t, http.MethodPatch, updateURL(fx, "bab", "titogo", "ha.go.tito.io"), |
| + token, `{"addrs":["127.0.0.1:7002"]}`) |
| + if status != http.StatusConflict { |
| + t.Fatalf("patch losing the race = %d %s, want 409", status, out) |
| + } |
| + fx.srv.cfg.Releases = fx.rel |
| + rec, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "ha.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Equal(rec.Addrs, []string{"127.0.0.1:7001"}) { |
| + t.Fatalf("the lost race still overwrote the record: %+v", rec) |
| + } |
| + if got := serial(t, fx); got != before { |
| + t.Fatalf("serial after a failed write = %d, want it unmoved from %d", got, before) |
| + } |
| +} |
| + |
| +// TestAPIInstanceListJoinsUpstreams: the list's third join (beside |
| +// staged_version and health) — per-upstream probe state, read from the box |
| +// snapshots the same way and with the same staleness rule, so `carlos |
| +// instances` can say WHICH upstream is down without a box session. Additive: |
| +// the key is absent entirely for a record no box is probing, which is every |
| +// single-upstream record in production today. |
| +func TestAPIInstanceListJoinsUpstreams(t *testing.T) { |
| + fx, token := instanceFixture(t, "code-inst-upstreams") |
| + enableInstances(t, fx, "titogo", titoManifest()) |
| + u := instancesURL(fx, "bab", "titogo") |
| + for _, h := range []string{"ha.go.tito.io", "plain.go.tito.io", "stale.go.tito.io"} { |
| + if code, _, out := rawReq(t, http.MethodPost, u, token, |
| + fmt.Sprintf(`{"host":%q,"region":"ie"}`, h)); code != http.StatusCreated { |
| + t.Fatalf("create %s = %d %s", h, code, out) |
| + } |
| + } |
| + since := time.Now().UTC().Add(-2 * time.Minute).Truncate(time.Second) |
| + fresh := status.CollectBox(nil, status.Probes{MachineID: "box1"}, time.Now()) |
| + fresh.Metrics = &status.BoxMetrics{Upstreams: map[string][]status.UpstreamState{ |
| + "ha.go.tito.io": { |
| + {Addr: "127.0.0.1:7001", Up: true, Since: since}, |
| + {Addr: "127.0.0.1:7002", Up: false, Since: since, Fails: 3, LastError: "dial refused"}, |
| + }, |
| + }} |
| + if err := status.WriteBox(context.Background(), fx.rel, fresh); err != nil { |
| + t.Fatal(err) |
| + } |
| + stale := status.CollectBox(nil, status.Probes{MachineID: "box2"}, time.Now().Add(-time.Hour)) |
| + stale.Metrics = &status.BoxMetrics{Upstreams: map[string][]status.UpstreamState{ |
| + "stale.go.tito.io": {{Addr: "127.0.0.1:7009", Up: true, Since: since}}, |
| + }} |
| + if err := status.WriteBox(context.Background(), fx.rel, stale); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + code, _, out := rawReq(t, http.MethodGet, u, token, "") |
| + if code != http.StatusOK { |
| + t.Fatalf("list = %d %s, want 200", code, out) |
| + } |
| + var list struct { |
| + Instances []struct { |
| + Host string `json:"host"` |
| + Upstreams []status.UpstreamState `json:"upstreams"` |
| + } `json:"instances"` |
| + } |
| + if err := json.Unmarshal([]byte(out), &list); err != nil { |
| + t.Fatalf("list body %s: %v", out, err) |
| + } |
| + got := map[string][]status.UpstreamState{} |
| + for _, r := range list.Instances { |
| + got[r.Host] = r.Upstreams |
| + } |
| + ha := got["ha.go.tito.io"] |
| + if len(ha) != 2 { |
| + t.Fatalf("ha upstreams = %+v, want both probed addrs", ha) |
| + } |
| + if ha[0].Addr != "127.0.0.1:7001" || !ha[0].Up { |
| + t.Errorf("first upstream = %+v, want 7001 up", ha[0]) |
| + } |
| + if ha[1].Addr != "127.0.0.1:7002" || ha[1].Up || ha[1].Fails != 3 || ha[1].LastError != "dial refused" { |
| + t.Errorf("second upstream = %+v, want 7002 down with its fail count and reason", ha[1]) |
| + } |
| + if len(got["plain.go.tito.io"]) != 0 { |
| + t.Errorf("an unprobed record carries upstreams: %+v", got["plain.go.tito.io"]) |
| + } |
| + // A box that stopped reporting drops OUT of the join rather than serving |
| + // its last-known belief as current — status.BoxStale, the same rule |
| + // staged_version votes with. |
| + if len(got["stale.go.tito.io"]) != 0 { |
| + t.Errorf("a stale box's probe table was joined: %+v", got["stale.go.tito.io"]) |
| + } |
| + // Absent, never present-and-empty: a client that never heard of this key |
| + // parses the record exactly as it always did. |
| + var wire struct { |
| + Instances []map[string]json.RawMessage `json:"instances"` |
| + } |
| + if err := json.Unmarshal([]byte(out), &wire); err != nil { |
| + t.Fatal(err) |
| + } |
| + for i, r := range list.Instances { |
| + _, present := wire.Instances[i]["upstreams"] |
| + if present != (r.Host == "ha.go.tito.io") { |
| + t.Errorf("%s upstreams key present = %v, want it only on the probed record", r.Host, present) |
| + } |
| + } |
| +} |
| + |
| // TestAPIInstanceDelete: deletion is idempotent (the caller's goal is "not |
| // there"), survives the manifest being retired, and still refuses a host |
| // that isn't one. |
diff --git a/internal/console/authcli.go b/internal/console/authcli.go| index b2c0a2c42..65b55da76 100644 |
| --- a/internal/console/authcli.go |
| +++ b/internal/console/authcli.go |
| @@ -277,6 +277,11 @@ func (s *Server) registerCLIRoutes() { |
| s.mux.HandleFunc("GET /api/cli/apps/{account}/{app}/logs", s.handleAPILogs) |
| s.mux.HandleFunc("POST /api/cli/apps/{account}/{app}/instances", s.handleAPIInstanceCreate) |
| s.mux.HandleFunc("GET /api/cli/apps/{account}/{app}/instances/{host}", s.handleAPIInstanceGet) |
| + // PATCH is the multi-upstream repoint (issue #152): the two fields of a |
| + // record — its addr set and its probe — that change while the instance |
| + // keeps serving. Same {host} wildcard, same gate, same byte-identical |
| + // 404 as its GET and DELETE siblings. |
| + s.mux.HandleFunc("PATCH /api/cli/apps/{account}/{app}/instances/{host}", s.handleAPIInstanceUpdate) |
| s.mux.HandleFunc("DELETE /api/cli/apps/{account}/{app}/instances/{host}", s.handleAPIInstanceDelete) |
| s.mux.HandleFunc("POST /api/cli/apps/{account}/{app}/instance-token", s.handleAPIInstanceToken) |
| // The config API (apiconfig.go) — `carlos env` and `carlos secrets` |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index d2c6655e1..53bd36893 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -571,23 +571,70 @@ func DeleteRecord(ctx context.Context, s release.Store, acct, app, host string) |
| return s.Delete(ctx, RecordKey(acct, app, host)) |
| } |
| +// UpdateRecord rewrites an EXISTING instance record, guarded by the etag the |
| +// caller read it under. Three writes now touch a record and each has its own |
| +// contract, deliberately: |
| +// |
| +// - CreateRecord is create-only ("" etag): a lost race is a permanent 409, |
| +// because two signups for one host must never silently merge. |
| +// - SetRecordRegion is an internal read-modify-write that retries once: it |
| +// stamps ONE field from a placement decision made elsewhere, so re-reading |
| +// and re-applying on a lost race changes nothing about what it meant. |
| +// - This one surfaces the conflict. Its caller is the console's member-facing |
| +// PATCH door, where the read and the write are one decision — a member sent |
| +// the upstream set they wanted GIVEN the record they were shown — so a |
| +// silent retry would re-apply that decision on top of a body nobody read, |
| +// which is the clobber PutManifestIfMatch's own doc comment records the |
| +// cost of. |
| +// |
| +// The record is marshaled here, by the same json.Marshal every other writer |
| +// uses: the caller hands over a whole provision.Record (already |
| +// ValidateShape'd) and never hand-builds the object body, so field ORDER and |
| +// omitempty behaviour — which the registry's joined addrs column compares |
| +// verbatim — cannot drift between the create path and this one. |
| +// |
| +// An empty etag is rejected rather than quietly becoming a create-only Put: |
| +// this verb is "rewrite what I read", and a caller that reached here with no |
| +// etag read nothing. |
| +func UpdateRecord(ctx context.Context, s release.Store, rec Record, etag string) error { |
| + if etag == "" { |
| + return fmt.Errorf("provision: update %s/%s/%s: no etag — an update must be guarded by the read it was decided from", |
| + rec.Account, rec.App, rec.Host) |
| + } |
| + body, err := json.Marshal(rec) |
| + if err != nil { |
| + return err |
| + } |
| + _, err = s.Put(ctx, RecordKey(rec.Account, rec.App, rec.Host), body, etag) |
| + return err |
| +} |
| + |
| // GetRecord reads and validates one instance record. A body whose own |
| // Host/App/Account disagree with the key path it was read from is refused |
| // rather than trusted — the same "never guessed at" discipline |
| // AccountManifest's sqid-in-path-and-body carries. |
| func GetRecord(ctx context.Context, s release.Store, acct, app, host string) (*Record, error) { |
| - raw, _, err := s.Get(ctx, RecordKey(acct, app, host)) |
| + rec, _, err := GetRecordETag(ctx, s, acct, app, host) |
| + return rec, err |
| +} |
| + |
| +// GetRecordETag is GetRecord carrying the etag its bytes were read under — |
| +// for a caller that must write back GUARDED against what it read |
| +// (UpdateRecord), not against whatever is there at write time. Same seam, |
| +// same reason, as GetManifestETag beside PutManifestIfMatch. |
| +func GetRecordETag(ctx context.Context, s release.Store, acct, app, host string) (*Record, string, error) { |
| + raw, etag, err := s.Get(ctx, RecordKey(acct, app, host)) |
| if err != nil { |
| - return nil, err |
| + return nil, "", err |
| } |
| var rec Record |
| if err := json.Unmarshal(raw, &rec); err != nil { |
| - return nil, fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err) |
| + return nil, "", fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err) |
| } |
| if rec.Account != acct || rec.App != app || rec.Host != host { |
| - return nil, fmt.Errorf("provision: record %s/%s/%s: record disagrees with its own path", acct, app, host) |
| + return nil, "", fmt.Errorf("provision: record %s/%s/%s: record disagrees with its own path", acct, app, host) |
| } |
| - return &rec, nil |
| + return &rec, etag, nil |
| } |
| // ListRecords lists an app's instance records, sorted by host, capped at |
diff --git a/internal/provision/provision_test.go b/internal/provision/provision_test.go| index 98449faeb..fe31bef8a 100644 |
| --- a/internal/provision/provision_test.go |
| +++ b/internal/provision/provision_test.go |
| @@ -258,6 +258,59 @@ func TestSetRecordRegion(t *testing.T) { |
| } |
| } |
| +// TestUpdateRecordGuardedByETag: the write-back half of the console's PATCH |
| +// door. Unlike CreateRecord (create-only, a lost race is a permanent 409) |
| +// and unlike SetRecordRegion (an internal read-modify-write that retries |
| +// once), this write is guarded by the etag the CALLER read the record under |
| +// and a lost race SURFACES — the member's request read a record, decided |
| +// what the new upstream set should be from it, and a silent retry would |
| +// re-apply that decision on top of a body it never saw. |
| +func TestUpdateRecordGuardedByETag(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := Record{ |
| + Schema: 1, Host: "acme.go.tito.io", Slug: "acme", App: "tito", Account: "bad", |
| + Kind: "service", Addr: "127.0.0.1:7000", Channel: "stable", |
| + CreatedAt: time.Now().UTC(), CreatedBy: "alice", |
| + } |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatalf("CreateRecord: %v", err) |
| + } |
| + |
| + got, etag, err := GetRecordETag(ctx, s, "bad", "tito", "acme.go.tito.io") |
| + if err != nil { |
| + t.Fatalf("GetRecordETag: %v", err) |
| + } |
| + if etag == "" { |
| + t.Fatal("GetRecordETag returned no etag — a guarded write-back has nothing to guard on") |
| + } |
| + if got.Addr != "127.0.0.1:7000" { |
| + t.Fatalf("GetRecordETag body = %+v", got) |
| + } |
| + |
| + got.Addr = "" |
| + got.Addrs = []string{"127.0.0.1:7001", "127.0.0.1:7002"} |
| + if err := UpdateRecord(ctx, s, *got, etag); err != nil { |
| + t.Fatalf("UpdateRecord: %v", err) |
| + } |
| + after, err := GetRecord(ctx, s, "bad", "tito", "acme.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(after.Addrs) != 2 || after.Addrs[0] != "127.0.0.1:7001" || after.Addrs[1] != "127.0.0.1:7002" { |
| + t.Fatalf("Addrs = %v, want the pair in order", after.Addrs) |
| + } |
| + if after.CreatedBy != rec.CreatedBy || !after.CreatedAt.Equal(rec.CreatedAt) || after.Channel != rec.Channel { |
| + t.Fatalf("UpdateRecord touched the record's identity/provenance: %+v", after) |
| + } |
| + |
| + // The same (now stale) etag again is a lost race, surfaced — never a |
| + // silent overwrite of whatever landed in between. |
| + if err := UpdateRecord(ctx, s, *got, etag); !errors.Is(err, release.ErrCASConflict) { |
| + t.Fatalf("UpdateRecord with a stale etag = %v, want release.ErrCASConflict", err) |
| + } |
| +} |
| + |
| func TestManifestRoundTrip(t *testing.T) { |
| ctx := context.Background() |
| s := release.NewDirStore(t.TempDir()) |
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index 2d79e7f37..96bb668a3 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -79,13 +79,29 @@ func routeFor(rec Record, dataDir, seed string) registry.Route { |
| rt.Socket = DefaultSocket(rec.Host) |
| rt.DBPath = DefaultDBPath(dataDir, rec.Host) |
| } |
| - // Addrs/Health are read from the record's EXPLICIT fields only, never |
| - // AddrsOrDefault/HealthOrDefault's Addr-folding — a legacy Addr-only |
| - // record must mint a byte-identical route to before these fields |
| - // existed (registry.Route.Addrs docs the AddrsOrDefault fold-back that |
| - // keeps single-upstream callers working off Addr alone). |
| + rt.Addrs, rt.Health = upstreamCols(rec) |
| + return rt |
| +} |
| + |
| +// upstreamCols renders a record's two upstream columns exactly as a minted |
| +// route carries them: the comma-joined addr set and the probe-config JSON. |
| +// |
| +// It exists as its own function because TWO paths now write these columns — |
| +// the mint (routeFor, above) and the owned-row converge (convergeOwned, which |
| +// calls registry.SetUpstreams) — and a second, independent derivation is how |
| +// a route that was minted correctly comes to be "converged" onto a different |
| +// value every pass, forever. One derivation, two callers. |
| +// |
| +// Addrs is read from the record's EXPLICIT field only, never |
| +// AddrsOrDefault's Addr-folding — a legacy Addr-only record must mint (and |
| +// converge to) a byte-identical route to before these fields existed |
| +// (registry.Route.Addrs documents the fold-back that keeps single-upstream |
| +// callers working off Addr alone). Health, by contrast, goes through |
| +// HealthOrDefault: a multi-upstream route with no stated probe MUST be |
| +// probed, or the edge has no way to tell which of its upstreams is dark. |
| +func upstreamCols(rec Record) (addrs, health string) { |
| if len(rec.Addrs) > 0 { |
| - rt.Addrs = strings.Join(rec.Addrs, ",") |
| + addrs = strings.Join(rec.Addrs, ",") |
| } |
| // WithDefaults, not the raw pointer: a record with explicit-but-partial |
| // probe config (e.g. only Path set) must still mint a route whose |
| @@ -98,13 +114,13 @@ func routeFor(rec Record, dataDir, seed string) registry.Route { |
| // Health is a struct of strings and ints — json.Marshal cannot fail |
| // on it (no channels, funcs, cyclic pointers, unsupported types). |
| // The err == nil guard exists to fail CLOSED on the impossible case |
| - // rather than panic: leaving rt.Health == "" would mean the route |
| + // rather than panic: leaving health == "" would mean the route |
| // mints unprobed, not that a bad value reaches the edge. |
| if b, err := json.Marshal(withDefaults); err == nil { |
| - rt.Health = string(b) |
| + health = string(b) |
| } |
| } |
| - return rt |
| + return addrs, health |
| } |
| // shapeMatches reports whether an existing route is the same ROUTE the |
| @@ -207,13 +223,24 @@ func shapeMismatch(existing, want registry.Route) string { |
| } |
| } |
| -// convergeOwned applies the record's Channel, Environment and StatusNamed onto |
| -// a route it owns — the columns the record is authoritative for once it owns |
| -// the row, whether ownership was just granted by adoption or was already held |
| -// by an earlier pass. Both the owned-route converge branch and the |
| -// adoption branch call this, so there is exactly one place these columns get |
| +// convergeOwned applies the record's Channel, Environment, StatusNamed and |
| +// UPSTREAM SET onto a route it owns — the columns the record is authoritative |
| +// for once it owns the row, whether ownership was just granted by adoption or |
| +// was already held by an earlier pass. Both the owned-route converge branch and |
| +// the adoption branch call this, so there is exactly one place these columns get |
| // written from a record, never two copies drifting apart. |
| // |
| +// Addrs/Health converge UNCONDITIONALLY on divergence — no "only when the |
| +// record states one" guard of the kind Unit carries — and that difference is |
| +// deliberate. Unit has a supported out-of-band writer (`carlos route -unit`, |
| +// issue #110) whose stamp a record that never mentioned the field must not |
| +// treat as drift. The upstream set has no such writer: the record IS the |
| +// desired state, the console's PATCH door is how it changes, and clearing the |
| +// set (PATCH with an empty addrs) has to reach the row or that door would move |
| +// the bucket and nothing else. Both values come from upstreamCols, the same |
| +// derivation the mint uses, so a route minted from a record cannot be |
| +// "converged" onto anything but what it already carries. |
| +// |
| // StatusNamed converges only when the record STATES it (design §3, the same |
| // tri-state Hibernate carries): absent means "this record does not speak to |
| // the status page", and stamping a derived false over an operator's `carlos |
| @@ -259,6 +286,22 @@ func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record, |
| log.Printf("provision: converged %s status-named %v -> %v", rec.Host, existing.StatusNamed, *rec.StatusNamed) |
| } |
| } |
| + // One UPDATE for both columns, because they are one fact: an addr set and |
| + // the probe that decides which of those addrs the edge will send traffic |
| + // to. Writing them separately would leave a window in which a route's new |
| + // upstreams are live under the OLD probe's path/expected-status — a probe |
| + // that can mark every new upstream dark. |
| + if addrs, health := upstreamCols(rec); existing.Addrs != addrs || existing.Health != health { |
| + if dryRun { |
| + log.Printf("provision: dry-run: would converge %s upstreams %q -> %q (probe %q -> %q)", |
| + rec.Host, existing.Addrs, addrs, existing.Health, health) |
| + } else if err := reg.SetUpstreams(rec.Host, addrs, health); err != nil { |
| + log.Printf("provision: converge upstreams on %s: %v", rec.Host, err) |
| + } else { |
| + log.Printf("provision: converged %s upstreams %q -> %q (probe %q -> %q)", |
| + rec.Host, existing.Addrs, addrs, existing.Health, health) |
| + } |
| + } |
| } |
| // Reconcile materializes bucket instance records into registry routes for |
diff --git a/internal/provision/reconcile_test.go b/internal/provision/reconcile_test.go| index 6e42af20e..59eeb6ca0 100644 |
| --- a/internal/provision/reconcile_test.go |
| +++ b/internal/provision/reconcile_test.go |
| @@ -590,6 +590,133 @@ func TestReconcileAbsentStatusNamedDoesNotStompOptIn(t *testing.T) { |
| } |
| } |
| +// haRecord is the multi-upstream fixture: a service route fronting two |
| +// processes, the shape the HA design (issue #152) added Addrs/Health for. |
| +func haRecord(sqid string) Record { |
| + return Record{Schema: 1, Host: "ha.go.tito.io", Slug: "ha", App: "tito", |
| + Account: sqid, Pool: "tito", Region: "ie", Channel: "stable", Kind: "service", |
| + Addrs: []string{"127.0.0.1:7001", "127.0.0.1:7002"}, CreatedAt: time.Now()} |
| +} |
| + |
| +// TestReconcileConvergesUpstreams: a record whose upstream set (or probe |
| +// config) changes AFTER the route was minted converges onto the owned row |
| +// within one pass, exactly as Channel and Environment do. |
| +// |
| +// This is the drift Task 7's review found: registry.SetUpstreams existed with |
| +// no non-test caller and the owned-row branch compared Kind/Addr/Unit/Socket/ |
| +// DBPath only, so a record that gained, lost or reordered an upstream kept |
| +// serving the set it was minted with forever — silently, since the drift log |
| +// never looked at those columns either. The console's PATCH door writes |
| +// exactly this record field, so without this converge that door would move |
| +// the bucket and nothing else. |
| +func TestReconcileConvergesUpstreams(t *testing.T) { |
| + s, sqid := setup(t) |
| + rec := haRecord(sqid) |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := regWith(t) |
| + cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := *mustGet(t, reg, "ha.go.tito.io") |
| + if before.Addrs != "127.0.0.1:7001,127.0.0.1:7002" { |
| + t.Fatalf("minted Addrs = %q, want the pair", before.Addrs) |
| + } |
| + |
| + // One upstream replaced: the row must follow the record. |
| + rec.Addrs = []string{"127.0.0.1:7001", "127.0.0.1:7003"} |
| + putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec)) |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + after := *mustGet(t, reg, "ha.go.tito.io") |
| + if after.Addrs != "127.0.0.1:7001,127.0.0.1:7003" { |
| + t.Errorf("Addrs = %q, want the record's new set — the record moved and the row did not", after.Addrs) |
| + } |
| + after.Addrs = before.Addrs |
| + if after != before { |
| + t.Errorf("the upstream converge touched more than Addrs: before %+v after %+v", before, after) |
| + } |
| + |
| + // ORDER is shape-significant (shapeMismatch compares the joined column |
| + // verbatim), so a reordered record is a converge, not a no-op. |
| + rec.Addrs = []string{"127.0.0.1:7003", "127.0.0.1:7001"} |
| + putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec)) |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, "ha.go.tito.io").Addrs; got != "127.0.0.1:7003,127.0.0.1:7001" { |
| + t.Errorf("Addrs = %q, want the record's order", got) |
| + } |
| + |
| + // The probe config converges through the same setter, and lands |
| + // fully-defaulted — an un-defaulted interval_ms of 0 reaching the edge |
| + // would be a tight-loop probe against a customer upstream. |
| + rec.Health = &Health{Path: "/healthz"} |
| + putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec)) |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + var probe Health |
| + if err := json.Unmarshal([]byte(mustGet(t, reg, "ha.go.tito.io").Health), &probe); err != nil { |
| + t.Fatalf("row Health is not probe JSON: %v", err) |
| + } |
| + if probe.Path != "/healthz" || probe.IntervalMS != 2000 || probe.Status != 200 { |
| + t.Errorf("converged probe = %+v, want /healthz with the platform defaults filled in", probe) |
| + } |
| + |
| + // A pass over an unchanged record writes nothing further: converge is |
| + // idempotent, not a write every tick. |
| + steady := *mustGet(t, reg, "ha.go.tito.io") |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if again := *mustGet(t, reg, "ha.go.tito.io"); again != steady { |
| + t.Errorf("a second pass over an unchanged record moved the row: %+v -> %+v", steady, again) |
| + } |
| +} |
| + |
| +// TestReconcileLeavesLegacySingleAddrUpstreamsAlone is the additivity guard |
| +// on the converge above: every record in a production bucket today is |
| +// single-Addr or socket-backed, carries no addrs and no probe, and must |
| +// converge to the same empty columns it has always had — never to a |
| +// one-element Addrs the fold-back (AddrsOrDefault) would produce. |
| +func TestReconcileLeavesLegacySingleAddrUpstreamsAlone(t *testing.T) { |
| + s, sqid := setup(t) |
| + for _, rec := range []Record{ |
| + acmeRecord(sqid), // socket-backed instance, no addr at all |
| + {Schema: 1, Host: "whole.go.tito.io", Slug: "whole", App: "tito", Account: sqid, |
| + Pool: "tito", Region: "ie", Channel: "stable", Kind: "service", |
| + Addr: "127.0.0.1:9443", CreatedAt: time.Now()}, |
| + } { |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + reg := regWith(t) |
| + cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := map[string]registry.Route{} |
| + for _, h := range []string{"acme.go.tito.io", "whole.go.tito.io"} { |
| + before[h] = *mustGet(t, reg, h) |
| + if before[h].Addrs != "" || before[h].Health != "" { |
| + t.Fatalf("%s minted with upstream columns set: %+v", h, before[h]) |
| + } |
| + } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + for h, want := range before { |
| + if got := *mustGet(t, reg, h); got != want { |
| + t.Errorf("%s moved on a converge pass: %+v -> %+v", h, want, got) |
| + } |
| + } |
| +} |
| + |
| // 5. Conflict. |
| func TestReconcileLeavesConflictAlone(t *testing.T) { |
| s, sqid := setup(t) |