console: self-serve mode with recovery mode, the paging arm and bumpSerialErr
selfServeMode is the switch, or recovery mode: anything the pass still owes (an open row, a pending release, owed effects or open record, an undelivered event, a reservation in active.json) found at start, or a start read that failed (fail closed). selfServeAdmits is the switch alone. armMovesPaging arms a claims-off console on the same conditions, and on a read failure. bumpSerial is bumpSerialErr, logged.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
6 files changed,
+209
−19
cmd/console/main.go+18 −5cmd/console/movesswitch_test.go+82 −4internal/console/selfserve_gate.go+20 −5internal/console/selfserve_pass.go+69 −0internal/console/serialbump.go+13 −4internal/console/server.go+7 −1
diff --git a/cmd/console/main.go b/cmd/console/main.go| index a6ce8575a..3016fd1ab 100644 |
| --- a/cmd/console/main.go |
| +++ b/cmd/console/main.go |
| @@ -494,7 +494,7 @@ func serve() { |
| } |
| // The move pass's pages (moves Plan 5 Task 7): the console's alert sinks, |
| // armed before the claim runtime binds the pass, so no pass runs unarmed. |
| - armMovesPaging(context.Background(), cfg, os.Getenv) |
| + armMovesPaging(context.Background(), cfg, st, os.Getenv) |
| if err := srv.ConfigureClaimRuntime(context.Background()); err != nil { |
| log.Fatalf("console: claim recovery: %v", err) |
| } |
| @@ -1478,10 +1478,23 @@ func applyMovesSwitch(cfg *console.Config, getenv func(string) string) { |
| } |
| // armMovesPaging arms alerting's sinks from CARLOS_ALERT_WEBHOOK and |
| -// CARLOS_ALERT_SNS_TOPIC_ARN for claim recovery and its optional move pass. |
| -// Claim loss pages run even when moves are disabled. |
| -func armMovesPaging(ctx context.Context, cfg console.Config, getenv func(string) string) bool { |
| - if !cfg.HostClaim { |
| +// CARLOS_ALERT_SNS_TOPIC_ARN for claim recovery and its optional move pass, |
| +// and for the self-serve moves pass. Claim loss pages run even when moves |
| +// are disabled. A claims-off console arms when self-serve mode's |
| +// conditions hold: the switch on, or anything the pass still owes in the |
| +// journal or the reservations (console.SelfServeEvidence). A read that |
| +// fails arms, logged: a pass that may have moves to page about never runs |
| +// unarmed. |
| +func armMovesPaging(ctx context.Context, cfg console.Config, st *console.Store, getenv func(string) string) bool { |
| + arm := cfg.HostClaim || cfg.SelfServeSwitchOn() |
| + if !arm { |
| + found, err := console.SelfServeEvidence(ctx, st, cfg.Releases) |
| + if err != nil { |
| + log.Printf("console: self-serve moves: reading the journal for paging: %v (paging armed)", err) |
| + } |
| + arm = found |
| + } |
| + if !arm { |
| return false |
| } |
| alerting.Arm(alerting.FromEnv(ctx, getenv, nil)) |
diff --git a/cmd/console/movesswitch_test.go b/cmd/console/movesswitch_test.go| index 48730c28d..fa334d5e8 100644 |
| --- a/cmd/console/movesswitch_test.go |
| +++ b/cmd/console/movesswitch_test.go |
| @@ -2,14 +2,19 @@ package main |
| import ( |
| "context" |
| + "encoding/json" |
| "go/ast" |
| "go/parser" |
| "go/token" |
| + "path/filepath" |
| "testing" |
| + "time" |
| "github.com/carlosframework/platform/internal/alerting" |
| "github.com/carlosframework/platform/internal/console" |
| "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| ) |
| // The Phase 1 moves switch reaches console.Config.MovesEnabled through |
| @@ -115,7 +120,7 @@ func TestArmMovesPagingWithClaims(t *testing.T) { |
| } |
| for _, tc := range []struct{ claim, moves, want bool }{{false, false, false}, {true, false, true}, {false, true, false}, {true, true, true}} { |
| cfg := console.Config{HostClaim: tc.claim, MovesEnabled: tc.moves} |
| - if got := armMovesPaging(context.Background(), cfg, env); got != tc.want { |
| + if got := armMovesPaging(context.Background(), cfg, pagingStore(t), env); got != tc.want { |
| t.Errorf("claim %v moves %v: armed %v, want %v", tc.claim, tc.moves, got, tc.want) |
| } |
| } |
| @@ -138,8 +143,8 @@ func TestArmMovesPagingWithClaims(t *testing.T) { |
| } |
| switch fun := call.Fun.(type) { |
| case *ast.Ident: |
| - if fun.Name == "armMovesPaging" && len(call.Args) == 3 { |
| - if sel, ok := call.Args[2].(*ast.SelectorExpr); ok && isIdent(sel.X, "os") && sel.Sel.Name == "Getenv" { |
| + if fun.Name == "armMovesPaging" && len(call.Args) == 4 && isIdent(call.Args[2], "st") { |
| + if sel, ok := call.Args[3].(*ast.SelectorExpr); ok && isIdent(sel.X, "os") && sel.Sel.Name == "Getenv" { |
| armed = call.Pos() |
| } |
| } |
| @@ -152,6 +157,79 @@ func TestArmMovesPagingWithClaims(t *testing.T) { |
| }) |
| } |
| if armed == 0 || configured == 0 || armed > configured { |
| - t.Fatalf("serve() must call armMovesPaging(ctx, cfg, os.Getenv) before ConfigureClaimRuntime (armed %v, configured %v)", armed, configured) |
| + t.Fatalf("serve() must call armMovesPaging(ctx, cfg, st, os.Getenv) before ConfigureClaimRuntime (armed %v, configured %v)", armed, configured) |
| + } |
| +} |
| + |
| +// pagingStore is an empty console store. |
| +func pagingStore(t *testing.T) *console.Store { |
| + t.Helper() |
| + st, err := console.Open(filepath.Join(t.TempDir(), "console.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { st.Close() }) |
| + return st |
| +} |
| + |
| +// A claims-off console pages from the self-serve moves pass whenever |
| +// self-serve mode's conditions hold (self-serve moves plan 2 Task 4): the |
| +// switch on, or anything the pass still owes in the journal or the |
| +// reservations; a read that fails arms too. |
| +func TestPagingArmedForSelfServe(t *testing.T) { |
| + env := func(k string) string { |
| + if k == "CARLOS_ALERT_WEBHOOK" { |
| + return "http://example.test/hook" |
| + } |
| + return "" |
| + } |
| + t.Cleanup(func() { alerting.Arm(nil) }) |
| + ctx := context.Background() |
| + arm := func(cfg console.Config, st *console.Store) bool { return armMovesPaging(ctx, cfg, st, env) } |
| + |
| + if arm(console.Config{}, pagingStore(t)) { |
| + t.Error("a claims-off console with the switch off and nothing owed armed paging") |
| + } |
| + if !arm(console.Config{SelfServeMoves: "offline"}, pagingStore(t)) { |
| + t.Error("the switch on did not arm paging") |
| + } |
| + if !arm(console.Config{SelfServeMoves: "offline", HostClaim: true}, pagingStore(t)) { |
| + t.Error("a host-claim console did not arm paging") |
| + } |
| + |
| + // An open row, with the switch off. |
| + st := pagingStore(t) |
| + now := time.Now() |
| + ok, _, err := st.AdmitSelfServeMove(console.SelfServeMove{ID: "mv_pagingpagingpagingpaging0", Account: "bab", App: "shop", |
| + Host: "shop.example.com", From: "ie", To: "uk", Participants: provision.Participants{ |
| + Source: []provision.Participant{{BoxID: "ie-1", Label: "ie-1", InstanceID: "i-1"}}, |
| + Target: provision.Participant{BoxID: "uk-1", Label: "uk-1", InstanceID: "i-2"}}, |
| + CreatedAt: now}) |
| + if err != nil || !ok { |
| + t.Fatal(ok, err) |
| + } |
| + if !arm(console.Config{}, st) { |
| + t.Error("an open self-serve row did not arm paging") |
| + } |
| + |
| + // A reservation, with the switch off and an empty journal. |
| + rel := release.NewDirStore(t.TempDir()) |
| + body, err := json.Marshal(moves.OfflineActiveBody(map[string]moves.OfflineReservation{ |
| + "ie-1": {Kind: moves.OfflineKindMove, ID: "mv_x", Host: "shop.example.com", Account: "bab", App: "shop", ReservedAt: now}})) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := rel.Put(ctx, moves.OfflineActiveKey, body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !arm(console.Config{Releases: rel}, pagingStore(t)) { |
| + t.Error("a reservation did not arm paging") |
| + } |
| + |
| + // A store that cannot be read arms, logged. |
| + closed := pagingStore(t) |
| + closed.Close() |
| + if !arm(console.Config{}, closed) { |
| + t.Error("a journal read failure did not arm paging") |
| } |
| } |
diff --git a/internal/console/selfserve_gate.go b/internal/console/selfserve_gate.go| index f4da2636e..4f0208dc7 100644 |
| --- a/internal/console/selfserve_gate.go |
| +++ b/internal/console/selfserve_gate.go |
| @@ -15,13 +15,28 @@ import ( |
| // else gets the move family's byte-identical 404 after authentication and |
| // before any move evidence is read. |
| +// SelfServeSwitchOn reports whether the self-serve moves switch is on for |
| +// this configuration: CARLOS_SELF_SERVE_MOVES=offline on a claims-off |
| +// console with Phase 1 off. |
| +func (c Config) SelfServeSwitchOn() bool { |
| + return c.SelfServeMoves == "offline" && !c.HostClaim && !c.MovesEnabled |
| +} |
| + |
| // selfServeMode reports whether the three move mutation doors are the |
| -// self-serve doors: the switch is "offline" on a claims-off console with |
| -// Phase 1 off. Outside it they are the legacy doors, byte for byte. Task 4 |
| -// extends this with recovery mode (an open self-serve row, a pending |
| -// release, an undelivered event or a reservation found at start). |
| +// self-serve doors: the switch is on, or the console is in recovery mode |
| +// (it found an open self-serve row, a pending release, an undelivered event |
| +// or a reservation at start; selfserve_pass.go). Recovery mode is |
| +// re-evaluated by the pass every tick and never turned off while any of |
| +// those exist, so moves in flight run to completion with the switch off. |
| +// Outside self-serve mode the doors are the legacy doors, byte for byte. |
| func (s *Server) selfServeMode() bool { |
| - return s.cfg.SelfServeMoves == "offline" && !s.cfg.HostClaim && !s.cfg.MovesEnabled |
| + return s.cfg.SelfServeSwitchOn() || s.selfServeRecovery.Load() |
| +} |
| + |
| +// selfServeAdmits reports whether a new self-serve move may be admitted (a |
| +// new prepare): the switch on. Recovery mode alone admits nothing. |
| +func (s *Server) selfServeAdmits() bool { |
| + return s.cfg.SelfServeSwitchOn() |
| } |
| // moveCapable reports whether p may ask for a self-serve move of acct's app: |
diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go| new file mode 100644 |
| index 000000000..71887aeb0 |
| --- /dev/null |
| +++ b/internal/console/selfserve_pass.go |
| @@ -0,0 +1,69 @@ |
| +package console |
| + |
| +// The self-serve moves pass (self-serve moves plan 2, Task 4; spec §7): |
| +// recovery mode, and the worker that steps the journal's rows. |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "os" |
| + |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// SelfServeEvidence reports whether anything puts a console in recovery |
| +// mode: an open self-serve row, a pending release, owed effects or open |
| +// record, an undelivered event (the store), or any reservation in |
| +// active.json (rel; nil reads no reservations). An error means it could |
| +// not tell; callers fail closed and treat it as found. |
| +func SelfServeEvidence(ctx context.Context, st *Store, rel release.Store) (bool, error) { |
| + pending, err := st.SelfServeRecoveryPending() |
| + if err != nil { |
| + return true, fmt.Errorf("the self-serve journal: %w", err) |
| + } |
| + if pending || rel == nil { |
| + return pending, nil |
| + } |
| + raw, _, err := rel.Get(ctx, moves.OfflineActiveKey) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return false, nil |
| + } |
| + if err != nil { |
| + return true, fmt.Errorf("the reservations: %w", err) |
| + } |
| + table, err := moves.ParseOfflineActive(raw) |
| + if err != nil { |
| + return true, fmt.Errorf("the reservations: %w", err) |
| + } |
| + return len(table) > 0, nil |
| +} |
| + |
| +// selfServeScanAtStart puts the console in recovery mode when it starts |
| +// with anything the pass still owes, or when it cannot tell (fail closed: |
| +// the pass runs and retries the reads each tick). A host-claim console has |
| +// no self-serve moves (the switch needs claims off), so it scans nothing. |
| +func (s *Server) selfServeScanAtStart(ctx context.Context) { |
| + if s.cfg.HostClaim { |
| + return |
| + } |
| + found, err := SelfServeEvidence(ctx, s.store, s.cfg.Releases) |
| + if err != nil { |
| + log.Printf("console: self-serve moves: reading at start: %v (recovery mode on; the pass retries each tick)", err) |
| + } |
| + s.selfServeRecovery.Store(found) |
| +} |
| + |
| +// selfServeReevaluate is the tick's re-evaluation of recovery mode: a read |
| +// that succeeds sets it to what it found, so it turns off only once none |
| +// of the evidence exists; a read that fails changes nothing. |
| +func (s *Server) selfServeReevaluate(ctx context.Context) { |
| + found, err := SelfServeEvidence(ctx, s.store, s.cfg.Releases) |
| + if err != nil { |
| + log.Printf("console: self-serve moves: %v (recovery mode unchanged; retrying next tick)", err) |
| + return |
| + } |
| + s.selfServeRecovery.Store(found) |
| +} |
diff --git a/internal/console/serialbump.go b/internal/console/serialbump.go| index c6be80555..a19614ec2 100644 |
| --- a/internal/console/serialbump.go |
| +++ b/internal/console/serialbump.go |
| @@ -29,12 +29,21 @@ import ( |
| // clamp still applies, measured from here: detached from cancellation, not |
| // from time. |
| func (s *Server) bumpSerial(ctx context.Context) { |
| + if err := s.bumpSerialErr(ctx); err != nil { |
| + log.Printf("console: serial bump: %v (boxes fall back to the hourly pass)", err) |
| + } |
| +} |
| + |
| +// bumpSerialErr is bumpSerial's bump with its error returned, for a caller |
| +// that must know whether the announcement landed (the self-serve moves job |
| +// records the record_seq it last announced and bumps again until one |
| +// does). The same detached, 2s-clamped bump; no bucket is nothing to |
| +// announce into, not a failure. |
| +func (s *Server) bumpSerialErr(ctx context.Context) error { |
| if s.cfg.Releases == nil { |
| - return // no bucket on this console: nothing to announce into |
| + return nil |
| } |
| ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second) |
| defer cancel() |
| - if err := release.BumpSerial(ctx, s.cfg.Releases, time.Now()); err != nil { |
| - log.Printf("console: serial bump: %v (boxes fall back to the hourly pass)", err) |
| - } |
| + return release.BumpSerial(ctx, s.cfg.Releases, time.Now()) |
| } |
diff --git a/internal/console/server.go b/internal/console/server.go| index 9f9edcbf8..3be110587 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -894,6 +894,9 @@ type Server struct { |
| selfServeLeaseMu sync.Mutex |
| selfServeLease *leaseToken |
| selfServeClock func() time.Time |
| + // selfServeRecovery is recovery mode (selfServeMode): set by the scan |
| + // at start, re-evaluated by every tick of the pass. |
| + selfServeRecovery atomic.Bool |
| // The console's move pass (movesrecovery.go): its clock (nil is |
| // time.Now; a TEST SEAM) and when each close was first seen due. |
| movesClock func() time.Time |
| @@ -1636,8 +1639,11 @@ func NewServer(cfg Config, st *Store) (*Server, error) { |
| if s.lifetimesGoverned() { |
| s.readCutoverAtStart(context.Background()) |
| } |
| - // The reclaim rule reads the self-serve rows from this store. |
| + // The reclaim rule reads the self-serve rows from this store, and the |
| + // console starts in recovery mode when the journal or the reservations |
| + // hold anything the pass still owes. |
| s.selfServeRows = st |
| + s.selfServeScanAtStart(context.Background()) |
| // The pre-accounts backfill: operators become owners of ops, every granted |
| // app name is claimed there, and non-operator grant holders are left out |
| // and REPORTED — a per-app grant was never authority over ops. Additive and |