amadan

carlos / platform Public

console: the self-serve moves pass skeleton

StartSelfServeMovesPass runs while the console is in self-serve mode,
each tick a mutating unit of plan 1's admission: hold the console lease
(renewed every 20 s during the tick), re-evaluate recovery mode from the
journal and active.json, deliver the undelivered events under the lease's
deadline, then one step per open row in order. selfServeStep is Task 7's
and leaves every row as it found it. Off on a host-claim console, which
the claim invariant's fixture and probe pin.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude f618773752f3de3810d6c0e99f68d5572bfea200 parent 6b3327a
5 files changed, +173 −0
  • cmd/console/main.go +4 −0
  • internal/console/claim_invariant_fixtures_test.go +1 −0
  • internal/console/claim_invariant_probes_worker_test.go +1 −0
  • internal/console/claim_workers.go +1 −0
  • internal/console/selfserve_pass.go +166 −0
diff --git a/cmd/console/main.go b/cmd/console/main.go
index 3016fd1ab..42905355d 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -514,6 +514,10 @@ func serve() {
// Same placement rule as the alert watcher: started here, never in
// NewServer, so constructing a Server stays goroutine-free.
go srv.RunClaimWorker(context.Background(), "StartClaimRecoveryPass")
+ // The self-serve moves pass (self-serve moves plan 2): runs while the
+ // console is in self-serve mode (the switch on, or recovery mode for
+ // moves found at start), under the console lease.
+ go srv.RunClaimWorker(context.Background(), "StartSelfServeMovesPass")
go srv.RunClaimWorker(context.Background(), "StartEconomicsPass")
// The infrastructure inventory pass writes infra/inventory.json for
// /fleet/architecture (internal/console/infrapass.go). Same placement
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go
index 97d08a452..062df4ce2 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -1860,6 +1860,7 @@ var claimFixtures = map[string]claimFixture{
"worker StartLifetimesCutoverWatch": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
"worker StartLifetimeRecoveryPass": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
"worker StartIncarnationMigrationPass": claimRead("off on a host-claim console, where lifetimes are not for it: it logs that and reads nothing"),
+ "worker StartSelfServeMovesPass": claimRead("off on a host-claim console, where self-serve moves are not for it: it logs that and reads nothing"),
"worker StartAlertWatcher": claimNonApp("box alert state under alerts/boxes and operator mail"),
"worker StartEconomicsPass": claimNonApp("daily usage census, statements and bills: deployment and account economics"),
"worker StartGeoipJobs": claimNonApp("GeoIP orphan sweep, D1 expiry and feed status under control/geoip/: the platform database, no app"),
diff --git a/internal/console/claim_invariant_probes_worker_test.go b/internal/console/claim_invariant_probes_worker_test.go
index 19e51823d..8558bf816 100644
--- a/internal/console/claim_invariant_probes_worker_test.go
+++ b/internal/console/claim_invariant_probes_worker_test.go
@@ -82,6 +82,7 @@ var claimWorkerProbes = map[string]claimProbe{
"worker StartLifetimesCutoverWatch": claimLogWorkerProbe("StartLifetimesCutoverWatch", nil, "lifetimes cutover watch", "host-claim console"),
"worker StartLifetimeRecoveryPass": claimLogWorkerProbe("StartLifetimeRecoveryPass", nil, "lifetime recovery pass", "host-claim console"),
"worker StartIncarnationMigrationPass": claimLogWorkerProbe("StartIncarnationMigrationPass", nil, "incarnation migration pass", "host-claim console"),
+ "worker StartSelfServeMovesPass": claimLogWorkerProbe("StartSelfServeMovesPass", nil, "self-serve moves pass", "host-claim console"),
"worker LogPlacementDiscrepancies": claimLogWorkerProbe("LogPlacementDiscrepancies", claimRegionalApp,
msgPlacementDiscrepancy("drift.go.tito.io", "blog", "uk")),
diff --git a/internal/console/claim_workers.go b/internal/console/claim_workers.go
index 5eb6eb8b3..2e2d5d47f 100644
--- a/internal/console/claim_workers.go
+++ b/internal/console/claim_workers.go
@@ -23,6 +23,7 @@ func (s *Server) registerClaimWorkers() {
"StartRelay": s.StartRelay,
"StartMailEventPush": s.StartMailEventPush,
"StartDoorKeyResign": s.StartDoorKeyResign,
+ "StartSelfServeMovesPass": s.StartSelfServeMovesPass,
"ReportApexSquats": s.ReportApexSquats,
"LogPlacementDiscrepancies": s.LogPlacementDiscrepancies,
"PublishMovesConsole": s.publishMovesConsoleWorker,
diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go
index 71887aeb0..927d651c3 100644
--- a/internal/console/selfserve_pass.go
+++ b/internal/console/selfserve_pass.go
@@ -5,10 +5,12 @@ package console
import (
"context"
+ "database/sql"
"errors"
"fmt"
"log"
"os"
+ "time"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/release"
@@ -67,3 +69,167 @@ func (s *Server) selfServeReevaluate(ctx context.Context) {
}
s.selfServeRecovery.Store(found)
}
+
+// selfServeTickInterval is the pass's period (spec §7: every 5 seconds). A
+// var so a test can shorten it.
+var selfServeTickInterval = 5 * time.Second
+
+// selfServeDrainBatch bounds the events one tick delivers.
+const selfServeDrainBatch = 100
+
+// StartSelfServeMovesPass runs the self-serve moves pass while the console
+// is in self-serve mode: every selfServeTickInterval, one iteration admitted
+// through plan 1's gate as a writer. With the switch off it stops once
+// recovery mode ends (nothing left to recover), and it never runs on a
+// console that started with nothing to recover. A host-claim console has no
+// self-serve moves: it logs that and returns.
+func (s *Server) StartSelfServeMovesPass(ctx context.Context) {
+ if s.cfg.HostClaim {
+ log.Printf("console: self-serve moves pass: off on a host-claim console, where self-serve moves are not for it")
+ return
+ }
+ t := time.NewTicker(selfServeTickInterval)
+ defer t.Stop()
+ for s.selfServeMode() {
+ s.selfServeIteration(ctx)
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ }
+ }
+}
+
+// selfServeIteration is one tick as a unit of plan 1's admission (mutating:
+// the pass writes records). It reports whether the tick ran.
+func (s *Server) selfServeIteration(ctx context.Context) bool {
+ return s.lifetimesIteration(ctx, "self-serve moves", true, false, s.selfServeTick)
+}
+
+// selfServeTick is one pass: hold the lease (acquired or renewed), the
+// reservation scan (recovery mode re-evaluated from the journal and
+// active.json), then, only while this console holds the lease, the event
+// delivery and one step per open row in order. The lease is renewed every
+// selfServeLeaseRenewEvery while the tick runs, and checked with a fresh
+// clock before each row.
+func (s *Server) selfServeTick(ctx context.Context) {
+ t := s.holdSelfServeLease(ctx)
+ s.selfServeReevaluate(ctx)
+ if t == nil {
+ return
+ }
+ stop := s.keepSelfServeLease(ctx, t)
+ defer stop()
+ s.drainSelfServeEvents(ctx, t)
+ rows, err := s.store.OpenSelfServeMoves()
+ if err != nil {
+ log.Printf("console: self-serve moves: the open rows: %v (retrying next tick)", err)
+ return
+ }
+ for i := range rows {
+ if !t.valid(s.selfServeNow()) {
+ log.Printf("console: self-serve moves: the console lease ran out mid-tick; the rest wait for the next")
+ return
+ }
+ if err := s.selfServeStep(ctx, t, &rows[i]); err != nil {
+ log.Printf("console: self-serve move %s (%s): %v", rows[i].ID, rows[i].Host, err)
+ }
+ }
+}
+
+// holdSelfServeLease is the tick's lease: the one held, renewed when a
+// renewal is due, or a fresh acquisition when none is held or it ran out.
+// Nil when another console holds it or it could not be read.
+func (s *Server) holdSelfServeLease(ctx context.Context) *leaseToken {
+ s.selfServeLeaseMu.Lock()
+ defer s.selfServeLeaseMu.Unlock()
+ now := s.selfServeNow()
+ if t := s.selfServeLease; t.valid(now) {
+ if !t.renewalDue(now) {
+ return t
+ }
+ err := t.renew(ctx, now)
+ if err == nil {
+ return t
+ }
+ if !errors.Is(err, errSelfServeLeaseLost) {
+ log.Printf("console: self-serve moves: renewing the console lease: %v", err)
+ if t.valid(s.selfServeNow()) {
+ return t
+ }
+ }
+ }
+ t, ok, err := s.acquireSelfServeLease(ctx, now)
+ if err != nil {
+ log.Printf("console: self-serve moves: the console lease: %v (the pass waits)", err)
+ }
+ if !ok {
+ if err == nil && s.selfServeLease != nil {
+ log.Printf("console: self-serve moves: another console holds the moves lease")
+ }
+ s.selfServeLease = nil
+ return nil
+ }
+ s.selfServeLease = t
+ return t
+}
+
+// keepSelfServeLease renews t every selfServeLeaseRenewEvery until the
+// returned stop is called, so a long tick keeps its lease.
+func (s *Server) keepSelfServeLease(ctx context.Context, t *leaseToken) (stop func()) {
+ done := make(chan struct{})
+ finished := make(chan struct{})
+ go func() {
+ defer close(finished)
+ tick := time.NewTicker(selfServeLeaseRenewEvery)
+ defer tick.Stop()
+ for {
+ select {
+ case <-done:
+ return
+ case <-ctx.Done():
+ return
+ case <-tick.C:
+ if err := t.renew(ctx, s.selfServeNow()); err != nil {
+ log.Printf("console: self-serve moves: renewing the console lease mid-tick: %v", err)
+ }
+ }
+ }
+ }()
+ return func() { close(done); <-finished }
+}
+
+// drainSelfServeEvents delivers the undelivered events, each as one audit
+// row in the move's app's activity, at most once (DeliverSelfServeEvent),
+// under a context bounded by the lease. A failed delivery stays undelivered
+// for the next tick; the others go on.
+func (s *Server) drainSelfServeEvents(ctx context.Context, t *leaseToken) {
+ evs, err := s.store.UnemittedSelfServeEvents(selfServeDrainBatch)
+ if err != nil {
+ log.Printf("console: self-serve moves: the undelivered events: %v (retrying next tick)", err)
+ return
+ }
+ for _, e := range evs {
+ if !t.valid(s.selfServeNow()) {
+ return
+ }
+ dctx, cancel := context.WithDeadline(ctx, t.deadline())
+ err := s.store.DeliverSelfServeEvent(dctx, e, func(tx *sql.Tx) error {
+ return auditActivityTx(tx, e.Account, e.App, "self-serve move", Activity{
+ Action: "instance-move", TargetType: "instance", Target: e.Host, Message: e.Phase, Detail: e.Reason})
+ })
+ cancel()
+ if err != nil {
+ log.Printf("console: self-serve move %s: delivering event %d: %v (retrying next tick)", e.ID, e.Seq, err)
+ }
+ }
+}
+
+// selfServeStep advances one row by at most one step. Task 7 writes it: the
+// row's deadlines, the judged record, reconciliation by the compatible
+// sets, the expiry action, abort and the phase function, every record write
+// through selfServeApply under selfServeWriteCtx. Until then it leaves
+// every row exactly as it found it.
+func (s *Server) selfServeStep(ctx context.Context, t *leaseToken, row *SelfServeMove) error {
+ return nil
+}