amadan

carlos / platform Public

console: the self-serve moves console lease and the record write guard

control/offline-moves/console.json, a 60 s lease taken by one conditional
write and renewed against its own ETag; a malformed lease is never
overwritten. provision.WithWriteGuard makes WriteRecord check a guard
immediately before every Put, and selfServeWriteCtx bounds a write's
context by the lease and guards each attempt with a fresh clock.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 026c43e67bbc5b67a80e3a77448c61df5121d0ad parent d66d055
4 files changed, +247 −1
  • internal/console/selfserve_lease.go +190 −0
  • internal/console/server.go +8 −0
  • internal/provision/record_sequence.go +18 −1
  • internal/provision/record_sequence_test.go +31 −0
diff --git a/internal/console/selfserve_lease.go b/internal/console/selfserve_lease.go
new file mode 100644
index 000000000..6a170895d
--- /dev/null
+++ b/internal/console/selfserve_lease.go
@@ -0,0 +1,190 @@
+package console
+
+// The console lease (self-serve moves plan 2, Task 4). One console runs the
+// self-serve moves pass: the one holding control/offline-moves/console.json,
+// a 60 s lease renewed every 20 s during a tick. The SQLite journal is the
+// console's, so a replacement console takes over by opening the same
+// database file once the lease has run out; the lease guards the day there
+// are two.
+//
+// Every record write the pass makes runs under selfServeWriteCtx: the
+// lease's deadline bounds its context, and a write guard checks the lease
+// with a fresh clock immediately before every Put attempt. A cancelled
+// context does not prove a write cannot commit; that is why every write is
+// a persisted request (Task 7).
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const (
+ selfServeLeaseKey = "control/offline-moves/console.json"
+ selfServeLeaseTTL = 60 * time.Second
+)
+
+// selfServeLeaseRenewEvery is how often a tick renews the lease it holds.
+// A var so a test can shorten it.
+var selfServeLeaseRenewEvery = 20 * time.Second
+
+// errSelfServeLeaseLost is a write or renewal refused because this console
+// does not hold the lease (it ran out, or another console took it).
+var errSelfServeLeaseLost = errors.New("console: self-serve moves: this console does not hold the moves lease")
+
+// consoleLease is console.json's body: which console holds the lease, and
+// until when.
+type consoleLease struct {
+ Console string `json:"console"`
+ Until time.Time `json:"until"`
+}
+
+// leaseToken is this console's hold on the lease: the object's ETag as last
+// written, and the time it runs out.
+type leaseToken struct {
+ s *Server
+ console string
+ mu sync.Mutex
+ etag string
+ until time.Time
+}
+
+// selfServeNow is the pass's clock (selfServeClock, a TEST SEAM, else
+// time.Now).
+func (s *Server) selfServeNow() time.Time {
+ if s.selfServeClock != nil {
+ return s.selfServeClock()
+ }
+ return time.Now()
+}
+
+// selfServeConsoleID names this console process in the lease: random, so
+// two consoles never share a name, and stable for the process.
+func (s *Server) selfServeConsoleID() string {
+ s.selfServeIDOnce.Do(func() {
+ var b [12]byte
+ if _, err := rand.Read(b[:]); err != nil {
+ panic(err) // crypto/rand does not fail on a supported platform
+ }
+ s.selfServeConsole = "console-" + hex.EncodeToString(b[:])
+ })
+ return s.selfServeConsole
+}
+
+// acquireSelfServeLease takes the lease at now when it is absent, has run
+// out, or is already this console's: one conditional write against the
+// ETag read (create-only when absent). Held by another console, or a lost
+// race, is not acquired and no error. An unreadable or malformed lease is
+// an error: it is never overwritten blind.
+func (s *Server) acquireSelfServeLease(ctx context.Context, now time.Time) (*leaseToken, bool, error) {
+ if s.cfg.Releases == nil {
+ return nil, false, errors.New("console: self-serve moves: no deployment bucket for the lease")
+ }
+ me := s.selfServeConsoleID()
+ raw, etag, err := s.cfg.Releases.Get(ctx, selfServeLeaseKey)
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ etag = ""
+ case err != nil:
+ return nil, false, err
+ default:
+ var cur consoleLease
+ if err := json.Unmarshal(raw, &cur); err != nil {
+ return nil, false, fmt.Errorf("%s: %w", selfServeLeaseKey, err)
+ }
+ if cur.Console == "" || cur.Until.IsZero() {
+ return nil, false, fmt.Errorf("%s: no console or no time", selfServeLeaseKey)
+ }
+ if cur.Console != me && now.Before(cur.Until) {
+ return nil, false, nil
+ }
+ }
+ t := &leaseToken{s: s, console: me}
+ if err := t.put(ctx, etag, now); err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ return nil, false, nil
+ }
+ return nil, false, err
+ }
+ return t, true, nil
+}
+
+// put writes the lease until now + TTL against ifMatch and records the new
+// ETag. The caller holds t.mu or owns t alone.
+func (t *leaseToken) put(ctx context.Context, ifMatch string, now time.Time) error {
+ until := now.Add(selfServeLeaseTTL).UTC()
+ body, err := json.Marshal(consoleLease{Console: t.console, Until: until})
+ if err != nil {
+ return err
+ }
+ tag, err := t.s.cfg.Releases.Put(ctx, selfServeLeaseKey, body, ifMatch)
+ if err != nil {
+ return err
+ }
+ t.etag, t.until = tag, until
+ return nil
+}
+
+// renew extends the lease to now + TTL with one conditional write against
+// the ETag this console last wrote. A lost race means another console took
+// it (or a renewal whose reply was lost moved the ETag): the token is no
+// longer valid and the error is errSelfServeLeaseLost; acquiring again
+// takes back a lease that is still this console's. Any other failure
+// leaves the token as it was, running out on its own.
+func (t *leaseToken) renew(ctx context.Context, now time.Time) error {
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ err := t.put(ctx, t.etag, now)
+ if errors.Is(err, release.ErrCASConflict) {
+ t.until = time.Time{}
+ return errSelfServeLeaseLost
+ }
+ return err
+}
+
+// valid reports whether the lease is still this console's at now.
+func (t *leaseToken) valid(now time.Time) bool {
+ if t == nil {
+ return false
+ }
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ return now.Before(t.until)
+}
+
+// deadline is when the lease runs out unless renewed.
+func (t *leaseToken) deadline() time.Time {
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ return t.until
+}
+
+// renewalDue reports whether a tick starting at now should renew: every
+// selfServeLeaseRenewEvery since the last write.
+func (t *leaseToken) renewalDue(now time.Time) bool {
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ return !now.Before(t.until.Add(-selfServeLeaseTTL).Add(selfServeLeaseRenewEvery))
+}
+
+// selfServeWriteCtx is the context every record write of the pass runs
+// under: bounded by the lease's deadline, with a write guard that refuses
+// any Put attempt once the lease is no longer valid by a fresh clock.
+func (s *Server) selfServeWriteCtx(ctx context.Context, t *leaseToken) (context.Context, context.CancelFunc) {
+ ctx, cancel := context.WithDeadline(ctx, t.deadline())
+ return provision.WithWriteGuard(ctx, func() error {
+ if !t.valid(s.selfServeNow()) {
+ return errSelfServeLeaseLost
+ }
+ return nil
+ }), cancel
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index 64ba53be2..9f9edcbf8 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -886,6 +886,14 @@ type Server struct {
// the boxes' cleanup reports (plan 4; nil settles no cleanup).
selfServeRows selfServeRowEvidence
cleanupReports cleanupReportEvidence
+ // The self-serve moves pass's console lease (selfserve_lease.go): this
+ // process's name in it, the lease while held, and the pass's clock
+ // (nil is time.Now; a TEST SEAM).
+ selfServeIDOnce sync.Once
+ selfServeConsole string
+ selfServeLeaseMu sync.Mutex
+ selfServeLease *leaseToken
+ selfServeClock func() time.Time
// 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
diff --git a/internal/provision/record_sequence.go b/internal/provision/record_sequence.go
index f8c307f18..b1cc71c51 100644
--- a/internal/provision/record_sequence.go
+++ b/internal/provision/record_sequence.go
@@ -99,6 +99,17 @@ func AdvanceRecordSequence(before, body []byte) ([]byte, error) {
return json.Marshal(fields)
}
+type writeGuardKey struct{}
+
+// WithWriteGuard makes WriteRecord call guard immediately before every Put
+// under ctx, and refuse the write with guard's error. The self-serve moves
+// pass binds every record write to its console lease this way: the lease is
+// checked with a fresh clock before each attempt, not only when the
+// context was made. A guard grants no write authority; it can only refuse.
+func WithWriteGuard(ctx context.Context, guard func() error) context.Context {
+ return context.WithValue(ctx, writeGuardKey{}, guard)
+}
+
// preparedRecord binds one synchronous Put to its bytes and CAS base. Store
// wrappers may call WriteRecord again without advancing twice. An inner
// claim boundary can introduce the counter and report the written value back
@@ -116,7 +127,8 @@ type preparedRecord struct {
// Legacy absent counters and no claim context retain the exact body bytes.
// A before marked by a migration (incarnation_migrating) is refused with
// ErrRecordMigrating unless ctx carries WithMigration of that marker; a
-// create-only write reads no before and is unaffected.
+// create-only write reads no before and is unaffected. A guard on ctx
+// (WithWriteGuard) runs immediately before the Put and can refuse it.
func WriteRecord(ctx context.Context, s release.Store, key string, before, body []byte, etag string) (string, uint64, error) {
if !IsRecordKey(key) {
return "", 0, ErrRecordSequence
@@ -153,6 +165,11 @@ func WriteRecord(ctx context.Context, s release.Store, key string, before, body
}
p.body, p.seq, p.sequenced = append([]byte(nil), body...), base+1, true
}
+ if guard, _ := ctx.Value(writeGuardKey{}).(func() error); guard != nil {
+ if err := guard(); err != nil {
+ return "", 0, err
+ }
+ }
tag, err := s.Put(ctx, key, body, etag)
return tag, p.seq, err
}
diff --git a/internal/provision/record_sequence_test.go b/internal/provision/record_sequence_test.go
index 59c019935..11bdef0ef 100644
--- a/internal/provision/record_sequence_test.go
+++ b/internal/provision/record_sequence_test.go
@@ -216,3 +216,34 @@ func TestTheMarkerIsTheOneDecodeRecordReads(t *testing.T) {
t.Fatalf("a marker that is not a string = %v, want ErrRecordSequence", err)
}
}
+
+// The write guard (self-serve moves plan 2, Task 4) runs immediately before
+// every record Put under its context, and its refusal stops the Put.
+func TestWriteRecordGuardRunsBeforeEveryPut(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ key := RecordKey("bab", "shop", markHost)
+ refuse := errors.New("the lease ran out")
+ calls := 0
+ allow := true
+ ctx := WithWriteGuard(context.Background(), func() error {
+ calls++
+ if !allow {
+ return refuse
+ }
+ return nil
+ })
+ body := []byte(`{"schema":1,"host":"` + markHost + `"}`)
+ etag, _, err := WriteRecord(ctx, s, key, []byte(`{}`), body, "")
+ if err != nil || calls != 1 {
+ t.Fatalf("first write = %v, %d guard calls", err, calls)
+ }
+ allow = false
+ next := []byte(`{"schema":1,"host":"` + markHost + `","app":"shop"}`)
+ if _, _, err := WriteRecord(ctx, s, key, body, next, etag); !errors.Is(err, refuse) || calls != 2 {
+ t.Fatalf("refused write = %v, %d guard calls", err, calls)
+ }
+ got, _, err := s.Get(context.Background(), key)
+ if err != nil || !bytes.Equal(got, body) {
+ t.Fatalf("the refused write reached the bucket: %s, %v", got, err)
+ }
+}