amadan

carlos / platform Public

self-serve pass: the evidence read's deadline covers the journal (fix round 2, item 5)

SelfServeRecoveryPending takes a context (QueryRowContext) and
SelfServeEvidence sets its 30 s bound before the journal read, so the
startup scan, the paging-arm read and every tick's re-evaluation give up at
the deadline when the store's one connection is held, failing closed into
recovery mode, logged. A startup:scan hook marks the scan's start.

Witness: TestStartupRecoveryScanBoundsTheJournalRead (NewServer with the
store's connection held by a transaction when the scan starts; then the
evidence read directly). RED: "mode false admits false after a journal
read that could not run".
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 36ca1bfc430d6d0866241e96c0da5007d526f6c3 parent c038dce
3 files changed, +64 −8
  • internal/console/selfserve_pass.go +6 −5
  • internal/console/selfserve_startup_test.go +53 −0
  • internal/console/selfserve_store.go +5 −3
diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go
index c0ac708b9..f2c5e9032 100644
--- a/internal/console/selfserve_pass.go
+++ b/internal/console/selfserve_pass.go
@@ -24,20 +24,20 @@ var selfServeEvidenceTimeout = 30 * time.Second
// 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. The reservations
-// read is bounded by selfServeEvidenceTimeout, so a stalled bucket cannot
+// not tell; callers fail closed and treat it as found. The journal read
+// and the reservations read are bounded by selfServeEvidenceTimeout, so a stalled bucket cannot
// hold the console's start or its paging arm (final review F12): a read
// that times out is an error like any other.
func SelfServeEvidence(ctx context.Context, st *Store, rel release.Store) (bool, error) {
- pending, err := st.SelfServeRecoveryPending()
+ ctx, cancel := context.WithTimeout(ctx, selfServeEvidenceTimeout)
+ defer cancel()
+ pending, err := st.SelfServeRecoveryPending(ctx)
if err != nil {
return true, fmt.Errorf("the self-serve journal: %w", err)
}
if pending || rel == nil {
return pending, nil
}
- ctx, cancel := context.WithTimeout(ctx, selfServeEvidenceTimeout)
- defer cancel()
raw, _, err := rel.Get(ctx, moves.OfflineActiveKey)
if errors.Is(err, os.ErrNotExist) {
return false, nil
@@ -64,6 +64,7 @@ func (s *Server) selfServeScanAtStart(ctx context.Context) {
}
ctx, cancel := context.WithTimeout(ctx, selfServeEvidenceTimeout)
defer cancel()
+ TestHooks.point("startup:scan")
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)
diff --git a/internal/console/selfserve_startup_test.go b/internal/console/selfserve_startup_test.go
index d1a9e92c3..2e3420fef 100644
--- a/internal/console/selfserve_startup_test.go
+++ b/internal/console/selfserve_startup_test.go
@@ -81,3 +81,56 @@ func TestStartupRecoveryScanIsBounded(t *testing.T) {
t.Fatalf("the evidence read over a stalled bucket = %v %v after %v; want found, an error, bounded", found, err, time.Since(start))
}
}
+
+// TestStartupRecoveryScanBoundsTheJournalRead (fix round 2, item 5): the
+// journal half of the evidence read is bounded too. With the console
+// store's one connection held when NewServer's recovery scan starts (a
+// SQLite read that cannot proceed), the scan gives up at its deadline and
+// the console starts in recovery mode (fail closed), logged; the
+// paging-arm read (SelfServeEvidence) returns at its deadline the same way.
+func TestStartupRecoveryScanBoundsTheJournalRead(t *testing.T) {
+ defer func(d time.Duration) { selfServeEvidenceTimeout = d }(selfServeEvidenceTimeout)
+ selfServeEvidenceTimeout = 200 * time.Millisecond
+ regPath, relStore, priv, pub := buildFixtures(t)
+ st, err := Open(filepath.Join(t.TempDir(), "console.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { st.Close() })
+ hold := func(d time.Duration) {
+ // The store's one connection, taken by a transaction for d.
+ tx, err := st.db.BeginTx(context.Background(), nil)
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ time.AfterFunc(d, func() { _ = tx.Rollback() })
+ }
+ TestHooks.Fault(func(point string) {
+ if point == "startup:scan" {
+ hold(1500 * time.Millisecond)
+ }
+ })
+ defer TestHooks.Fault(nil)
+ var logs strings.Builder
+ restore := captureLogTo(&logs)
+ s, err := NewServer(Config{Origin: "http://127.0.0.1", Keymail: "http://keymail.invalid", Registry: regPath, Releases: relStore,
+ ReleasePub: pub, Key: priv, AppScheme: "http"}, st)
+ restore()
+ TestHooks.Fault(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !s.selfServeMode() || s.selfServeAdmits() {
+ t.Fatalf("mode %v admits %v after a journal read that could not run; want recovery mode (fail closed)", s.selfServeMode(), s.selfServeAdmits())
+ }
+ if !strings.Contains(logs.String(), "recovery mode on") {
+ t.Fatalf("the timed-out journal read was not logged: %q", logs.String())
+ }
+ hold(1500 * time.Millisecond)
+ start := time.Now()
+ found, err := SelfServeEvidence(context.Background(), st, relStore)
+ if took := time.Since(start); !found || err == nil || took > time.Second {
+ t.Fatalf("the evidence read with the journal blocked = %v %v after %v; want found, an error, within the bound", found, err, took)
+ }
+}
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
index ef779333b..3be7c5829 100644
--- a/internal/console/selfserve_store.go
+++ b/internal/console/selfserve_store.go
@@ -566,10 +566,12 @@ func (s *Store) OpenSelfServeMoves() ([]SelfServeMove, error) {
// SelfServeRecoveryPending reports whether the journal holds anything the
// pass still owes: an open row, a pending release, owed effects or open
// record, or an undelivered event. Any of them puts the console in
-// recovery mode (selfServeMode).
-func (s *Store) SelfServeRecoveryPending() (bool, error) {
+// recovery mode (selfServeMode). ctx bounds the read, the wait for the
+// store's connection included (fix round 2: the evidence read's deadline
+// covers the journal).
+func (s *Store) SelfServeRecoveryPending(ctx context.Context) (bool, error) {
var n int
- err := s.db.QueryRow(`SELECT (SELECT COUNT(*) FROM self_serve_moves WHERE ` + selfServeOpenWhere + `)
+ err := s.db.QueryRowContext(ctx, `SELECT (SELECT COUNT(*) FROM self_serve_moves WHERE `+selfServeOpenWhere+`)
+ (SELECT COUNT(*) FROM self_serve_events WHERE emitted = 0)`).Scan(&n)
return n > 0, err
}