| index a99cc3f89..e7befef93 100644 |
| --- a/internal/console/apisystem.go |
| +++ b/internal/console/apisystem.go |
| @@ -65,6 +65,20 @@ func (j *systemJob) finish(err error) { |
| j.mu.Unlock() |
| } |
| |
| +func (j *systemJob) done() bool { |
| + j.mu.Lock() |
| + defer j.mu.Unlock() |
| + return j.Done |
| +} |
| + |
| +// snapshot copies the three published fields under the lock, so the caller |
| +// can serialize them without holding it across a socket write. |
| +func (j *systemJob) snapshot() (steps []string, done bool, errMsg string) { |
| + j.mu.Lock() |
| + defer j.mu.Unlock() |
| + return append([]string(nil), j.Steps...), j.Done, j.Err |
| +} |
| + |
| // systemUpload is the one parked local build: `carlos system update --file` |
| // PUTs the bytes, then names the handle on the update request. There is no |
| // timestamp on it and no TTL sweep, because there is only ever ONE — the next |
| @@ -105,6 +119,18 @@ func (s *Server) apiSystemOwner(w http.ResponseWriter, r *http.Request) (string, |
| // subject is the deployment itself — so there is no *Account to pass and a |
| // nil would panic. The ops account is the right home for the row: its owners |
| // are the only callers that ever reach these doors. |
| +// |
| +// KNOW WHAT THIS IS AND IS NOT. These rows are keyed (OpsAccountID, app |
| +// "platform"), and NO SURFACE READS THEM TODAY: Store.Events is asked for a |
| +// real app name by the activity tab, and "platform" is not one. That is |
| +// deliberate, not an oversight — the row is the DURABLE record of who moved |
| +// the platform pointer and when, written where a reader can be added later |
| +// (an operator activity view, `carlos system log`) without a migration. |
| +// Until such a reader exists this is a write-only table, and the answer to |
| +// "who promoted this?" comes from the pointer's own Audit list, which |
| +// PromoteSystem signs into the bucket. Do not delete these rows on the |
| +// grounds that nothing reads them; do not present them as an audit trail an |
| +// operator can consult, because they cannot. |
| func (s *Server) auditSystem(addr, action, detail string) { |
| if err := s.store.RecordEvent(OpsAccountID, "platform", addr, action, detail, time.Now()); err != nil { |
| log.Printf("console: audit %s platform: %v", action, err) |
| @@ -126,36 +152,85 @@ func (s *Server) systemBucket(w http.ResponseWriter) bool { |
| return true |
| } |
| |
| +// systemFeed is the release feed this console fetches from: cfg.SystemFeed |
| +// when it is configured, the vendor feed otherwise. |
| +// |
| +// "Configured" means ALL THREE strings, not just one. A feed's three fields |
| +// are a single fact — where to ask, where to download, and the key the |
| +// checksum list must verify under — and any subset of them is not a feed at |
| +// all: a DownloadBase with the vendor's PubKey would fetch one publisher's |
| +// binaries and check them against another's signature, which cannot ever |
| +// succeed and would read as "the release is corrupt" rather than as "this |
| +// console is misconfigured". So a partial override is refused OUT LOUD and |
| +// falls back, rather than being half-honoured. |
| func (s *Server) systemFeed() publicrelease.Feed { |
| - if s.cfg.SystemFeed.DownloadBase == "" { |
| + f := s.cfg.SystemFeed |
| + set := 0 |
| + for _, v := range []string{f.LatestURL, f.DownloadBase, f.PubKey} { |
| + if v != "" { |
| + set++ |
| + } |
| + } |
| + switch set { |
| + case 0: |
| + return publicrelease.Default() |
| + case 3: |
| + return f |
| + default: |
| + log.Printf("console: SystemFeed is only partly configured (latest_url=%t download_base=%t pub_key=%t) — "+ |
| + "a feed is all three or none, so this console is using the vendor feed instead", |
| + f.LatestURL != "", f.DownloadBase != "", f.PubKey != "") |
| return publicrelease.Default() |
| } |
| - return s.cfg.SystemFeed |
| } |
| |
| // systemBoxes reads every heartbeat: the platform prefix and each fleet's. |
| -func (s *Server) systemBoxes(ctx context.Context) ([]status.Box, error) { |
| +// |
| +// strict is what a fleet the console cannot enumerate MEANS, and the two |
| +// callers need opposite answers. The promote (systemPlatforms) decides from |
| +// this list which platforms to mirror, and a fleet it cannot see is a fleet |
| +// whose architecture it would silently omit — so strict refuses rather than |
| +// guesses. The status read only DESCRIBES, and an operator asking what is |
| +// running is better served by the boxes that did answer plus an honest |
| +// "partial" flag than by a blank page. The second return is that flag: true |
| +// means some fleet's boxes are missing from the slice. |
| +func (s *Server) systemBoxes(ctx context.Context, strict bool) ([]status.Box, bool, error) { |
| boxes, err := status.LoadBoxes(ctx, s.cfg.Releases) |
| if err != nil { |
| - return nil, err |
| + return nil, false, err |
| } |
| reg, err := s.openPassRegistry() |
| if err != nil { |
| - return boxes, nil // no registry: platform boxes only |
| + return s.systemFleetsUnreadable(boxes, strict, err) |
| } |
| fleets, err := reg.ListFleets() |
| if err != nil { |
| - return boxes, nil |
| + return s.systemFleetsUnreadable(boxes, strict, err) |
| } |
| + partial := false |
| for _, f := range fleets { |
| fb, err := status.LoadBoxes(ctx, release.NewPrefixStore(s.cfg.Releases, "fleets/"+f.Name+"/")) |
| if err != nil { |
| + if strict { |
| + return nil, false, fmt.Errorf("fleet %s: cannot read its box heartbeats, refusing to guess which platforms to mirror: %w", f.Name, err) |
| + } |
| log.Printf("console: api system: fleet %s boxes: %v", f.Name, err) |
| + partial = true |
| continue |
| } |
| boxes = append(boxes, fb...) |
| } |
| - return boxes, nil |
| + return boxes, partial, nil |
| +} |
| + |
| +// systemFleetsUnreadable is the shared answer to "this console cannot say |
| +// what fleets exist": a refusal under strict, a logged degradation otherwise. |
| +func (s *Server) systemFleetsUnreadable(boxes []status.Box, strict bool, err error) ([]status.Box, bool, error) { |
| + if strict { |
| + return nil, false, fmt.Errorf("cannot list fleets, refusing to guess which platforms to mirror: %w", err) |
| + } |
| + log.Printf("console: api system: list fleets: %v — reporting the platform's own boxes only", err) |
| + return boxes, true, nil |
| } |
| |
| type systemUpdateRequest struct { |
| @@ -206,31 +281,67 @@ func (s *Server) handleAPISystemUpdate(w http.ResponseWriter, r *http.Request) { |
| } |
| spread = d |
| } |
| - id, job := s.newSystemJob() |
| + id, job, busy := s.startSystemJob(in.Component, in.Channel) |
| + if busy != "" { |
| + writeJSON(w, http.StatusConflict, map[string]string{ |
| + "error": "a promote of " + in.Component + "/" + in.Channel + " is already running; watch it or wait for it to finish", |
| + "job": busy, |
| + }) |
| + return |
| + } |
| go s.runSystemUpdate(context.WithoutCancel(r.Context()), job, addr, in, spread) |
| writeJSON(w, http.StatusOK, map[string]string{"job": id}) |
| } |
| |
| -// newSystemJob mints a job id and its record, sweeping jobs older than the |
| -// TTL on the way past — the registry is process memory, and an operator who |
| -// promoted an hour ago is not still polling. |
| -func (s *Server) newSystemJob() (string, *systemJob) { |
| - b := make([]byte, 8) |
| - rand.Read(b) |
| - id := hex.EncodeToString(b) |
| - job := &systemJob{at: time.Now()} |
| +// startSystemJob claims the ONE in-flight promote slot for a channel and |
| +// mints the job that holds it, or reports the id of the job already there. |
| +// |
| +// One at a time per channel, and the reason is memory as much as |
| +// correctness: a promote holds a whole release asset in the heap while it |
| +// verifies and mirrors it, so N concurrent promotes are N copies, and the |
| +// door is reachable by an ops owner with a script. Serializing per channel |
| +// also removes the CAS race two promotes of different versions into the same |
| +// channel would otherwise run — one of them would lose at PromoteSystem's |
| +// Put and report a conflict the operator did not cause. |
| +// |
| +// It sweeps FINISHED jobs past the TTL on the way past. Keyed on Done rather |
| +// than age: a promote that has legitimately been running for over an hour |
| +// (a slow feed, a large asset) must not have its record swept out from under |
| +// the operator watching it, which is exactly the moment they need it. |
| +func (s *Server) startSystemJob(component, channel string) (id string, job *systemJob, busy string) { |
| s.systemMu.Lock() |
| defer s.systemMu.Unlock() |
| + if running, ok := s.systemRunning[component+"/"+channel]; ok { |
| + return "", nil, running |
| + } |
| + b := make([]byte, 8) |
| + rand.Read(b) |
| + id = hex.EncodeToString(b) |
| + job = &systemJob{at: time.Now()} |
| if s.systemJobs == nil { |
| s.systemJobs = map[string]*systemJob{} |
| } |
| + if s.systemRunning == nil { |
| + s.systemRunning = map[string]string{} |
| + } |
| for k, j := range s.systemJobs { |
| - if time.Since(j.at) > systemJobTTL { |
| + if j.done() && time.Since(j.at) > systemJobTTL { |
| delete(s.systemJobs, k) |
| } |
| } |
| s.systemJobs[id] = job |
| - return id, job |
| + s.systemRunning[component+"/"+channel] = id |
| + return id, job, "" |
| +} |
| + |
| +// releaseSystemChannel hands the slot back. It is called BEFORE the job is |
| +// marked done, so an operator who polls a job to completion and immediately |
| +// promotes again is never told the channel is busy by the job they just |
| +// watched finish. |
| +func (s *Server) releaseSystemChannel(component, channel string) { |
| + s.systemMu.Lock() |
| + delete(s.systemRunning, component+"/"+channel) |
| + s.systemMu.Unlock() |
| } |
| |
| // runSystemUpdate is the promote: resolve, ingest (or take the upload), |
| @@ -298,13 +409,18 @@ func (s *Server) runSystemUpdate(ctx context.Context, job *systemJob, actor stri |
| job.step("serial bumped: boxes on %s start their spread", in.Channel) |
| return nil |
| }() |
| + // Hand the channel back BEFORE marking the job done — see |
| + // releaseSystemChannel. |
| + s.releaseSystemChannel(in.Component, in.Channel) |
| job.finish(err) |
| } |
| |
| // systemPlatforms: the distinct platforms the fleet reports; linux-arm64 |
| -// when nothing has reported the field yet. |
| +// when nothing has reported the field yet. STRICT — a fleet this console |
| +// cannot enumerate stops the promote, because a platform silently left out |
| +// of the mirror is a fleet of boxes that will never adopt. |
| func (s *Server) systemPlatforms(ctx context.Context) ([]string, error) { |
| - boxes, err := s.systemBoxes(ctx) |
| + boxes, _, err := s.systemBoxes(ctx, true) |
| if err != nil { |
| return nil, err |
| } |
| @@ -329,29 +445,70 @@ func (s *Server) systemPlatforms(ctx context.Context) ([]string, error) { |
| // different sha → refuse (versions are immutable, spec §3). |
| func (s *Server) mirrorSystemArtifact(ctx context.Context, job *systemJob, version, platform string, data []byte, sha string) error { |
| key := release.SystemArtifactKey(version, platform) |
| - existing, _, err := s.cfg.Releases.Get(ctx, key) |
| - switch { |
| - case err == nil: |
| - sum := sha256.Sum256(existing) |
| - if hex.EncodeToString(sum[:]) == sha { |
| - job.step("%s already mirrored", key) |
| - return nil |
| - } |
| - return fmt.Errorf("%s exists with different bytes — versions are immutable; pick a new version", key) |
| - case errors.Is(err, os.ErrNotExist): |
| - default: |
| + switch already, err := s.systemArtifactMatches(ctx, key, sha); { |
| + case err != nil: |
| return err |
| + case already: |
| + job.step("%s already mirrored", key) |
| + return nil |
| } |
| // Create-only: the empty ifMatch IS the CAS, so a concurrent promote of |
| // the same version loses rather than overwrites a byte a box may already |
| - // have fetched. |
| + // have fetched. Losing that race is not a failure — the two channels' |
| + // promotes of one version want the same bytes at the same key — so |
| + // re-read and let the comparison above decide: identical is done, and |
| + // different is the immutability refusal it always was. |
| if _, err := s.cfg.Releases.Put(ctx, key, data, ""); err != nil { |
| - return err |
| + if !errors.Is(err, release.ErrCASConflict) { |
| + return err |
| + } |
| + // The re-read answers every half: identical bytes means the other |
| + // promote wrote what this one would have, different bytes comes back |
| + // as the immutability refusal, and absent means the key was taken and |
| + // released while this promote looked away. |
| + already, cerr := s.systemArtifactMatches(ctx, key, sha) |
| + if cerr != nil { |
| + return cerr |
| + } |
| + if !already { |
| + return fmt.Errorf("%s: another writer holds the key but it is not there now — re-run, every step is idempotent", key) |
| + } |
| + job.step("%s mirrored concurrently", key) |
| + return nil |
| } |
| job.step("mirrored %s", key) |
| return nil |
| } |
| |
| +// systemArtifactMatches answers "is sha already the object at key?" — |
| +// absent is (false, nil), present and different is the immutability refusal, |
| +// present and equal is (true, nil). |
| +// |
| +// The comparison STREAMS the object already in the bucket through a hasher |
| +// rather than Getting it. A promote already holds one whole asset in the |
| +// heap (the verified download, waiting to be Put); buffering the existing |
| +// object beside it would make the peak two copies of a platform binary per |
| +// in-flight promote, for no purpose beyond computing one sha256. |
| +func (s *Server) systemArtifactMatches(ctx context.Context, key, sha string) (bool, error) { |
| + rc, _, err := release.OpenObject(ctx, s.cfg.Releases, key) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + return false, nil |
| + case err != nil: |
| + return false, err |
| + } |
| + h := sha256.New() |
| + _, err = io.Copy(h, rc) |
| + rc.Close() |
| + if err != nil { |
| + return false, fmt.Errorf("%s: reading the mirrored object to compare it: %w", key, err) |
| + } |
| + if hex.EncodeToString(h.Sum(nil)) == sha { |
| + return true, nil |
| + } |
| + return false, fmt.Errorf("%s exists with different bytes — versions are immutable; pick a new version", key) |
| +} |
| + |
| func (s *Server) handleAPISystemJob(w http.ResponseWriter, r *http.Request) { |
| if _, ok := s.apiSystemOwner(w, r); !ok { |
| return |
| @@ -363,9 +520,8 @@ func (s *Server) handleAPISystemJob(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusNotFound, "no such job — the console may have restarted; re-run, every step is idempotent") |
| return |
| } |
| - job.mu.Lock() |
| - defer job.mu.Unlock() |
| - writeJSON(w, http.StatusOK, map[string]any{"steps": append([]string(nil), job.Steps...), "done": job.Done, "error": job.Err}) |
| + steps, done, errMsg := job.snapshot() |
| + writeJSON(w, http.StatusOK, map[string]any{"steps": steps, "done": done, "error": errMsg}) |
| } |
| |
| // handleAPISystemUpload parks one local build in memory for the job that |
| @@ -457,9 +613,20 @@ func (s *Server) handleAPISystemRollback(w http.ResponseWriter, r *http.Request) |
| if !readJSONStrictLimit(w, r, &in, 1<<10) { |
| return |
| } |
| + // The same two checks the update door makes, and for the same reason: |
| + // both arguments become part of an object key, and a door that validates |
| + // what it writes but not what it rewinds is a door with two standards. |
| if in.Component == "" { |
| in.Component = release.SystemComponentRouter |
| } |
| + if in.Component != release.SystemComponentRouter { |
| + jsonError(w, http.StatusBadRequest, "unknown component "+in.Component) |
| + return |
| + } |
| + if in.Channel != release.RungCanary && in.Channel != release.RungStable { |
| + jsonError(w, http.StatusBadRequest, "channel must be canary or stable") |
| + return |
| + } |
| p, err := release.RollbackSystem(r.Context(), s.cfg.Releases, release.RootSigner(s.cfg.Key), in.Component, in.Channel, addr) |
| switch { |
| case errors.Is(err, release.ErrNoSystemHistory): |
| @@ -491,15 +658,25 @@ func (s *Server) handleAPISystemStatus(w http.ResponseWriter, r *http.Request) { |
| targets := map[string]string{} |
| for _, ch := range []string{release.RungCanary, release.RungStable} { |
| p, _, err := release.GetSystemPointer(ctx, s.cfg.Releases, s.cfg.ReleasePub, release.SystemComponentRouter, ch) |
| - if err != nil { |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + // null means exactly one thing: this channel has never been |
| + // promoted. It must not also mean "there IS a pointer and this |
| + // console cannot verify it" — a tampered or wrongly-signed |
| + // pointer reading as "never promoted" is the one failure an |
| + // operator most needs to see, shown as the one thing they would |
| + // ignore. |
| pointers[ch] = nil |
| - continue |
| + case err != nil: |
| + log.Printf("console: api system status: %s pointer: %v", ch, err) |
| + pointers[ch] = map[string]any{"error": err.Error()} |
| + default: |
| + pointers[ch] = map[string]any{"version": p.Current.Version, "promoted_at": p.PromotedAt, "spread": p.Spread, "source": p.Current.Source, |
| + "history": len(p.History)} |
| + targets[ch] = p.Current.Version |
| } |
| - pointers[ch] = map[string]any{"version": p.Current.Version, "promoted_at": p.PromotedAt, "spread": p.Spread, "source": p.Current.Source, |
| - "history": len(p.History)} |
| - targets[ch] = p.Current.Version |
| } |
| - boxes, err := s.systemBoxes(ctx) |
| + boxes, partial, err := s.systemBoxes(ctx, false) |
| if err != nil { |
| jsonError(w, http.StatusBadGateway, "box heartbeats unreadable: "+err.Error()) |
| return |
| @@ -523,5 +700,9 @@ func (s *Server) handleAPISystemStatus(w http.ResponseWriter, r *http.Request) { |
| rows = append(rows, row) |
| } |
| sort.Slice(rows, func(i, j int) bool { return rows[i]["label"].(string) < rows[j]["label"].(string) }) |
| - writeJSON(w, http.StatusOK, map[string]any{"pointers": pointers, "boxes": rows}) |
| + // partial says the box list is INCOMPLETE — some fleet's heartbeats |
| + // could not be read. Always present, so a reader never has to tell a |
| + // missing key from a false one, and so "no boxes on that fleet" and |
| + // "could not ask that fleet" are different answers here. |
| + writeJSON(w, http.StatusOK, map[string]any{"pointers": pointers, "boxes": rows, "partial": partial}) |
| } |
amadan