| new file mode 100644 |
| index 000000000..1a8ee33aa |
| --- /dev/null |
| +++ b/internal/console/offlinereserve.go |
| @@ -0,0 +1,438 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "net/http" |
| + "os" |
| + "sort" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Per-box reservations (self-serve moves plan 2 Task 3; spec §7.1). |
| +// control/offline-moves/active.json is a table of reservations keyed by box |
| +// id, one entry per participating box of every operation from its |
| +// reservation to its release (moves.OfflineActive). An operation reserves |
| +// before it writes its own durable state (the record's first move state), |
| +// and holds the reservation until the move clears. The legacy fields are |
| +// the table's projection, so a console that predates the table still reads |
| +// one deployment-wide lock. |
| +// |
| +// Rollout and rollback. An old console cannot be reintroduced while any |
| +// reservation exists that the old console did not create: its release |
| +// deletes the whole object when the projection names its own move |
| +// (TestOldConsoleReleaseErasesATable). The supported sequence is roll the |
| +// consoles, then turn the switch on; rollback is switch off, then wait until |
| +// every incompatible operation ends (every self-serve row terminal and |
| +// released, every cleanup reservation gone, every operator reservation |
| +// created under the table form released), then the old console. |
| + |
| +// offlineSelection is a move's participants as the door or the job chose |
| +// them (Task 5's offlineRegionChecks produces it for self-serve; the |
| +// operator start and stage doors build one from offlineSelect's fleet |
| +// today, offlineOperatorSelection). Sentinel is set when any participant's |
| +// InstanceID is empty: such a selection reserves every box. |
| +type offlineSelection struct { |
| + From, To string |
| + Fleet offlineFleet |
| + Source []provision.Participant |
| + Target provision.Participant |
| + Stage provision.StageTarget |
| + ReturnOf string |
| + Sentinel bool |
| +} |
| + |
| +// reservationKeys are the keys the selection reserves: every participant's |
| +// BoxID, or the sentinel alone for a sentinel selection. A participant |
| +// without an InstanceID or a BoxID makes it a sentinel selection whatever |
| +// Sentinel says, so a box nobody can name is never left unreserved. |
| +func (sel offlineSelection) reservationKeys() []string { |
| + sentinel := []string{moves.OfflineSentinel} |
| + if sel.Sentinel { |
| + return sentinel |
| + } |
| + set := map[string]bool{} |
| + for _, p := range append(append([]provision.Participant(nil), sel.Source...), sel.Target) { |
| + if p.BoxID == "" || p.InstanceID == "" { |
| + return sentinel |
| + } |
| + set[p.BoxID] = true |
| + } |
| + keys := make([]string, 0, len(set)) |
| + for k := range set { |
| + keys = append(keys, k) |
| + } |
| + sort.Strings(keys) |
| + return keys |
| +} |
| + |
| +// offlineOperatorSelection is the operator start and stage doors' |
| +// selection: the source region's boxes and the one target box, as |
| +// offlineSelect found them, labelled and named by the box id each edges file |
| +// publishes. The operator path pins no InstanceID until Task 5's pinning |
| +// lands, so every operator selection is a sentinel selection: one operator |
| +// move at a time across the deployment, as before the table. |
| +func offlineOperatorSelection(f offlineFleet, from, to string, st provision.StageTarget) offlineSelection { |
| + sel := offlineSelection{From: from, To: to, Fleet: f, Stage: st} |
| + for _, b := range f.boxesAt(from) { |
| + sel.Source = append(sel.Source, provision.Participant{BoxID: b.BoxID, Label: b.Box}) |
| + } |
| + if tgt := f.boxesAt(to); len(tgt) > 0 { |
| + sel.Target = provision.Participant{BoxID: tgt[0].BoxID, Label: tgt[0].Box} |
| + } |
| + sel.Sentinel = true |
| + return sel |
| +} |
| + |
| +// selfServeRowEvidence is what reclaim rules (c) and (d) read from the |
| +// self-serve rows (Task 4's tables): MoveTerminalOrAbsent is true when no |
| +// row has the move id, or its row is terminal with release_pending 0; |
| +// HostHasIssuedRequest is true when a row for the host has request_issued |
| +// set. A nil Server.selfServeRows means no rows exist, so a reservation is |
| +// reclaimed on rules (a) and (b) alone, as an operator reservation is. |
| +type selfServeRowEvidence interface { |
| + MoveTerminalOrAbsent(id string) (bool, error) |
| + HostHasIssuedRequest(acct, app, host string) (bool, error) |
| +} |
| + |
| +// cleanupReportEvidence is what a cleanup reservation's reclaim reads from |
| +// its owning box's status report (plan 4): CleanupSettled is true when |
| +// status/boxes/<label>.json, written within two status intervals by the box |
| +// whose Metrics.BoxID is boxID, lists id terminal with no root request and |
| +// no release pending, or carries a complete Retirements block that does not |
| +// list it. Neither the report fields nor their reader exist before plan 4 |
| +// and Task 6, so a nil Server.cleanupReports means no status can settle a |
| +// cleanup: a cleanup reservation is never reclaimed until then. |
| +type cleanupReportEvidence interface { |
| + CleanupSettled(ctx context.Context, boxID, id string) (bool, error) |
| +} |
| + |
| +// The refusals. busy and unreadable are the sentences the one-move lock |
| +// answered; mismatch is the job's (it aborts on it). |
| +func offlineBusy(host string) *offlineRefusal { |
| + return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s is moving offline. move one instance at a time", host)} |
| +} |
| + |
| +var ( |
| + offlineReserveUnreadable = &offlineRefusal{status: http.StatusBadGateway, msg: "could not read the instance"} |
| + offlineReserveMismatch = &offlineRefusal{status: http.StatusConflict, msg: "the reservation does not match this move's participants"} |
| +) |
| + |
| +// offlineOp is one operation's identity in the table. |
| +type offlineOp struct{ kind, id string } |
| + |
| +func opOf(e moves.OfflineReservation) offlineOp { return offlineOp{e.Kind, e.ID} } |
| + |
| +// reserveBoxes reserves sel's boxes for r's operation, in this order: |
| +// |
| +// 1. read active.json (absent is the empty table; a legacy lock is the |
| +// sentinel entry, moves.ParseOfflineActive, normalised into the table |
| +// on the first write); |
| +// 2. validate the whole object: any malformed entry is unreadable; |
| +// 3. the own retry: entries with r's (Kind, ID) are accepted when they |
| +// cover exactly sel's keys (nothing is written and their reserved_at |
| +// stays); a strict subset is a partial reservation whose missing keys |
| +// are acquired below in the same write; any key outside sel's is the |
| +// mismatch refusal; |
| +// 4. with the switch off (cfg.SelfServeMoves != "offline") every entry of |
| +// another operation conflicts: one operation at a time, today's |
| +// deployment-wide semantics; |
| +// 5. otherwise an entry of another operation conflicts when it holds a |
| +// key sel needs, holds the sentinel, or sel is the sentinel; |
| +// a conflicting operation is busy unless it is reclaimable |
| +// (offlineReclaimable), and a reclaimed operation loses all its entries; |
| +// 6. one conditional Put of the table with every missing key and the |
| +// projection. A lost race re-reads once and starts again at 2, then |
| +// answers busy. |
| +// |
| +// The caller holds no app or hostname lock: a reclaim takes the holder's |
| +// (s.lockApp, then withHostname) around its fence only. |
| +func (s *Server) reserveBoxes(ctx context.Context, sel offlineSelection, r moves.OfflineReservation) *offlineRefusal { |
| + if r.ReservedAt.IsZero() { |
| + r.ReservedAt = time.Now().UTC() |
| + } |
| + want := sel.reservationKeys() |
| + wanted := map[string]bool{} |
| + for _, k := range want { |
| + wanted[k] = true |
| + } |
| + mine := opOf(r) |
| + var lastHolder string |
| + for attempt := 0; attempt < 2; attempt++ { |
| + raw, etag, err := s.cfg.Releases.Get(ctx, moves.OfflineActiveKey) |
| + table := map[string]moves.OfflineReservation{} |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + etag = "" |
| + case err != nil: |
| + log.Printf("console: offline move %s: reading the reservations: %v", r.Host, err) |
| + return offlineReserveUnreadable |
| + default: |
| + if table, err = moves.ParseOfflineActive(raw); err != nil { |
| + log.Printf("console: offline move %s: %v", r.Host, err) |
| + return offlineReserveUnreadable |
| + } |
| + } |
| + if p := moves.OfflineActiveBody(table); p.Host != "" { |
| + lastHolder = p.Host |
| + } |
| + |
| + // (3) The own retry. |
| + own := map[string]moves.OfflineReservation{} |
| + for k, e := range table { |
| + if opOf(e) == mine { |
| + if !wanted[k] { |
| + return offlineReserveMismatch |
| + } |
| + own[k] = e |
| + } |
| + } |
| + if len(own) == len(want) { |
| + return nil |
| + } |
| + entry := r |
| + for _, e := range own { |
| + // A partial reservation completes at its own reserved_at. |
| + entry = e |
| + break |
| + } |
| + |
| + // (4) and (5) The operations in the way. |
| + sentinelSel := len(want) == 1 && want[0] == moves.OfflineSentinel |
| + conflicts := map[offlineOp]map[string]moves.OfflineReservation{} |
| + for k, e := range table { |
| + op := opOf(e) |
| + if op == mine { |
| + continue |
| + } |
| + if s.cfg.SelfServeMoves != "offline" || sentinelSel || k == moves.OfflineSentinel || wanted[k] { |
| + if conflicts[op] == nil { |
| + conflicts[op] = map[string]moves.OfflineReservation{} |
| + } |
| + conflicts[op][k] = e |
| + } |
| + } |
| + if ref := s.offlineInTheWay(ctx, conflicts); ref != nil { |
| + return ref |
| + } |
| + |
| + // (6) One write. |
| + next := map[string]moves.OfflineReservation{} |
| + for k, e := range table { |
| + if conflicts[opOf(e)] == nil { |
| + next[k] = e |
| + } |
| + } |
| + for _, k := range want { |
| + if _, ok := next[k]; !ok { |
| + next[k] = entry |
| + } |
| + } |
| + body, err := json.Marshal(moves.OfflineActiveBody(next)) |
| + if err != nil { |
| + return offlineReserveUnreadable |
| + } |
| + _, err = s.cfg.Releases.Put(ctx, moves.OfflineActiveKey, body, etag) |
| + if err == nil { |
| + return nil |
| + } |
| + if !errors.Is(err, release.ErrCASConflict) { |
| + log.Printf("console: offline move %s: writing the reservations: %v", r.Host, err) |
| + return offlineReserveUnreadable |
| + } |
| + } |
| + if lastHolder == "" { |
| + return offlineReserveUnreadable |
| + } |
| + return offlineBusy(lastHolder) |
| +} |
| + |
| +// offlineInTheWay answers busy, naming the oldest holder, unless every |
| +// conflicting operation is reclaimable. A young operation is busy without |
| +// any reclaim being attempted, so nothing is fenced while one is. |
| +func (s *Server) offlineInTheWay(ctx context.Context, conflicts map[offlineOp]map[string]moves.OfflineReservation) *offlineRefusal { |
| + type held struct { |
| + op offlineOp |
| + entries map[string]moves.OfflineReservation |
| + first moves.OfflineReservation |
| + } |
| + var ops []held |
| + for op, entries := range conflicts { |
| + var first moves.OfflineReservation |
| + for _, e := range entries { |
| + if first.ReservedAt.IsZero() || e.ReservedAt.Before(first.ReservedAt) { |
| + first = e |
| + } |
| + } |
| + ops = append(ops, held{op, entries, first}) |
| + } |
| + sort.Slice(ops, func(i, j int) bool { |
| + a, b := ops[i].first, ops[j].first |
| + if !a.ReservedAt.Equal(b.ReservedAt) { |
| + return a.ReservedAt.Before(b.ReservedAt) |
| + } |
| + return ops[i].op.kind+"/"+ops[i].op.id < ops[j].op.kind+"/"+ops[j].op.id |
| + }) |
| + for _, h := range ops { |
| + if time.Since(h.first.ReservedAt) < offlineLockGrace { |
| + return offlineBusy(h.first.Host) |
| + } |
| + } |
| + for _, h := range ops { |
| + if !s.offlineReclaimable(ctx, h.entries) { |
| + return offlineBusy(h.first.Host) |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// offlineReclaimable is the reclaim rule for one operation's entries, all |
| +// older than offlineLockGrace (the caller checked). Positive evidence only: |
| +// any read failure, malformed body, refusal or doubt keeps the reservation. |
| +// |
| +// A move: (a) the holder's record is fenced (fenceRecord, under the |
| +// holder's s.lockApp then withHostname, and admitted) and the fence answers |
| +// fenced or absent; (b) the fenced observation carries no move with the id; |
| +// (c) the move's self-serve row is absent or terminal with release_pending |
| +// 0; (d) no row for the host has request_issued set. Before the fence a |
| +// plain judged read that already shows the move keeps the reservation with |
| +// nothing written, so a live move is never fenced by a busy check. The fence |
| +// is what makes the negative read conclusive: a delayed queue or operator |
| +// start write that commits before it is in the fenced observation, and one |
| +// that commits after it fails its CAS (the fence advanced record_seq and the |
| +// ETag), so a paused door that resumes after a reclaim cannot write. |
| +// |
| +// A cleanup: the owning box's status settles it (cleanupReportEvidence). |
| +func (s *Server) offlineReclaimable(ctx context.Context, entries map[string]moves.OfflineReservation) bool { |
| + for key, e := range entries { |
| + switch e.Kind { |
| + case moves.OfflineKindMove: |
| + return s.offlineMoveReclaimable(ctx, e) |
| + case moves.OfflineKindCleanup: |
| + if s.cleanupReports == nil { |
| + return false |
| + } |
| + ok, err := s.cleanupReports.CleanupSettled(ctx, key, e.ID) |
| + if err != nil { |
| + log.Printf("console: offline move: cleanup %s on %s: status: %v (kept)", e.ID, key, err) |
| + return false |
| + } |
| + if !ok { |
| + return false |
| + } |
| + default: |
| + return false |
| + } |
| + } |
| + return len(entries) > 0 |
| +} |
| + |
| +func (s *Server) offlineMoveReclaimable(ctx context.Context, e moves.OfflineReservation) bool { |
| + carries := func(rec *provision.Record) bool { return rec != nil && rec.Move != nil && rec.Move.ID == e.ID } |
| + pre := s.observeRecord(ctx, e.Account, e.App, e.Host) |
| + switch { |
| + case pre.Verdict == provision.LifetimeUnreadable, pre.Verdict == provision.LifetimeCorrupt: |
| + log.Printf("console: offline move: reservation %s of %s: the record could not be read (kept): %v", e.ID, e.Host, pre.Err) |
| + return false |
| + case pre.Present && carries(pre.Rec): |
| + return false |
| + } |
| + // LOCK ORDER: the holder's app lock, then its hostname exclusion (the |
| + // lock table in lifetimes.go), held around the fence only. reserveBoxes' |
| + // caller holds neither. |
| + var ( |
| + res fenceResult |
| + obs provision.Observation |
| + ferr error |
| + ) |
| + func() { |
| + defer s.lockAppHost(e.Account, e.App, e.Host)() |
| + if ferr = s.admit(e.Host, ""); ferr != nil { |
| + return |
| + } |
| + res, obs, ferr = s.fenceRecord(ctx, e.Account, e.App, e.Host, fenceOwner{}) |
| + }() |
| + if ferr != nil { |
| + log.Printf("console: offline move: reservation %s of %s: not fenced (kept): %v", e.ID, e.Host, ferr) |
| + return false |
| + } |
| + switch { |
| + case res == fenceAbsent: |
| + case res == fenced && obs.Rec != nil && obs.Verdict != provision.LifetimeCorrupt && !carries(obs.Rec): |
| + default: |
| + return false |
| + } |
| + if s.selfServeRows == nil { |
| + return true |
| + } |
| + done, err := s.selfServeRows.MoveTerminalOrAbsent(e.ID) |
| + if err != nil || !done { |
| + if err != nil { |
| + log.Printf("console: offline move: reservation %s: its row: %v (kept)", e.ID, err) |
| + } |
| + return false |
| + } |
| + issued, err := s.selfServeRows.HostHasIssuedRequest(e.Account, e.App, e.Host) |
| + if err != nil { |
| + log.Printf("console: offline move: reservation %s: the rows of %s: %v (kept)", e.ID, e.Host, err) |
| + return false |
| + } |
| + return !issued |
| +} |
| + |
| +// releaseBoxes removes every entry of the (kind, id) operation and rewrites |
| +// the projection, or deletes the object when nothing is left, each under the |
| +// ETag it read (a lost race reads again, three rounds). Absence, of the |
| +// object or of the operation's entries, is success; a malformed table is an |
| +// error and is left as it is. |
| +func (s *Server) releaseBoxes(ctx context.Context, kind, id string) error { |
| + op := offlineOp{kind, id} |
| + for round := 0; round < 3; round++ { |
| + raw, etag, err := s.cfg.Releases.Get(ctx, moves.OfflineActiveKey) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil |
| + } |
| + if err != nil { |
| + return fmt.Errorf("console: releasing %s %s: %w", kind, id, err) |
| + } |
| + table, err := moves.ParseOfflineActive(raw) |
| + if err != nil { |
| + return fmt.Errorf("console: releasing %s %s: %w", kind, id, err) |
| + } |
| + next := map[string]moves.OfflineReservation{} |
| + for k, e := range table { |
| + if opOf(e) != op { |
| + next[k] = e |
| + } |
| + } |
| + if len(next) == len(table) { |
| + return nil |
| + } |
| + if len(next) == 0 { |
| + err = release.DeleteIfMatch(ctx, s.cfg.Releases, moves.OfflineActiveKey, etag) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil |
| + } |
| + } else { |
| + var body []byte |
| + if body, err = json.Marshal(moves.OfflineActiveBody(next)); err != nil { |
| + return err |
| + } |
| + _, err = s.cfg.Releases.Put(ctx, moves.OfflineActiveKey, body, etag) |
| + } |
| + if err == nil { |
| + return nil |
| + } |
| + if !errors.Is(err, release.ErrCASConflict) { |
| + return fmt.Errorf("console: releasing %s %s: %w", kind, id, err) |
| + } |
| + } |
| + return fmt.Errorf("console: releasing %s %s: lost 3 rounds", kind, id) |
| +} |
amadan