amadan

carlos / platform Public

self-serve pass: the journal read is abandoned at its deadline (fix round 3, item 5)

QueryRowContext bounds only the pool wait: the SQLite driver prepares
before it installs interruption and its lock retry waits on an
uncancellable mutex. journalPending runs the journal read in its own
goroutine and SelfServeEvidence abandons it at the 30 s deadline, failing
closed into recovery mode (logged by the caller); the late answer lands in
a buffered channel and is discarded. The startup scan, every tick's
re-evaluation and (through the start scan's cached answer) the paging arm
share it.

Witness: TestStartupRecoveryScanBoundsTheJournalRead, three cases: a real
SQLite write lock (BEGIN IMMEDIATE on a second connection to the file,
held past the bound), which a WAL reader does not wait for, so the scan
answers within the bound from the journal as it is; the store's one
pooled connection held (fail closed at the deadline); and a read blocked
inside the driver where no context reaches (evidence:journal hook): fail
closed at the deadline. RED (the last): "the start scan took 1.505s, past
its bound 200ms".
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9ee2cf307f469aae4771d7b7593394d32d996598 parent 2940933
2 files changed, +140 −43
  • internal/console/selfserve_pass.go +27 −1
  • internal/console/selfserve_startup_test.go +113 −42
diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go
index 2f3d1735e..f0490646d 100644
--- a/internal/console/selfserve_pass.go
+++ b/internal/console/selfserve_pass.go
@@ -31,7 +31,7 @@ var selfServeEvidenceTimeout = 30 * time.Second
func SelfServeEvidence(ctx context.Context, st *Store, rel release.Store) (bool, error) {
ctx, cancel := context.WithTimeout(ctx, selfServeEvidenceTimeout)
defer cancel()
- pending, err := st.SelfServeRecoveryPending(ctx)
+ pending, err := journalPending(ctx, st)
if err != nil {
return true, fmt.Errorf("the self-serve journal: %w", err)
}
@@ -52,6 +52,31 @@ func SelfServeEvidence(ctx context.Context, st *Store, rel release.Store) (bool,
return len(table) > 0, nil
}
+// journalPending is the journal half of the evidence, bounded at the API
+// (fix round 3): the read runs in its own goroutine and is abandoned at
+// ctx's deadline, because a context does not reach every wait inside the
+// SQLite driver (it prepares a statement before it installs interruption,
+// and its lock retry waits on a mutex). An abandoned read's late result is
+// discarded; the caller fails closed.
+func journalPending(ctx context.Context, st *Store) (bool, error) {
+ type answer struct {
+ pending bool
+ err error
+ }
+ ch := make(chan answer, 1) // the late answer lands here and is dropped
+ go func() {
+ TestHooks.point("evidence:journal")
+ p, err := st.SelfServeRecoveryPending(ctx)
+ ch <- answer{p, err}
+ }()
+ select {
+ case a := <-ch:
+ return a.pending, a.err
+ case <-ctx.Done():
+ return true, ctx.Err()
+ }
+}
+
// 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
@@ -66,6 +91,7 @@ func (s *Server) selfServeScanAtStart(ctx context.Context) {
defer cancel()
TestHooks.point("startup:scan")
found, err := SelfServeEvidence(ctx, s.store, s.cfg.Releases)
+ TestHooks.point("startup:scanned")
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 2e3420fef..79ac4c42a 100644
--- a/internal/console/selfserve_startup_test.go
+++ b/internal/console/selfserve_startup_test.go
@@ -2,6 +2,8 @@ package console
import (
"context"
+ "database/sql"
+ "errors"
"path/filepath"
"strings"
"testing"
@@ -82,55 +84,124 @@ func TestStartupRecoveryScanIsBounded(t *testing.T) {
}
}
-// 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.
+// TestStartupRecoveryScanBoundsTheJournalRead (fix round 2, item 5; fix
+// round 3): the journal half of the evidence read is bounded at the API,
+// whatever happens inside the driver.
+//
+// - A real SQLite write lock held on the database file by a second
+// connection past the bound (BEGIN IMMEDIATE): the console's journal is
+// in WAL mode, where a reader never waits for a writer, so the scan
+// answers within the bound from the journal as it is (nothing owed);
+// NewServer's later writes wait for the lock, as any writer does.
+// - The store's one pooled connection held: the scan gives up at its
+// deadline and the console starts in recovery mode (fail closed).
+// - A read that blocks inside the driver where no context reaches it
+// (the driver prepares before it installs interruption, and its lock
+// retry waits on a mutex): the scan abandons it at the deadline, fails
+// closed into recovery mode, logged, and the late result is discarded.
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)
+ const hold = 1500 * time.Millisecond
+ type scan struct {
+ srv *Server
+ logs string
+ scanAt, doneAt time.Time
}
- 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)
+ boot := func(t *testing.T, at func(point string, st *Store, path string)) scan {
+ t.Helper()
+ regPath, relStore, priv, pub := buildFixtures(t)
+ path := filepath.Join(t.TempDir(), "console.db")
+ st, err := Open(path)
if err != nil {
- t.Error(err)
- return
+ t.Fatal(err)
}
- time.AfterFunc(d, func() { _ = tx.Rollback() })
- }
- TestHooks.Fault(func(point string) {
- if point == "startup:scan" {
- hold(1500 * time.Millisecond)
+ t.Cleanup(func() { time.Sleep(hold); st.Close() }) // the late read finishes first
+ var sc scan
+ TestHooks.Fault(func(point string) {
+ switch point {
+ case "startup:scan":
+ sc.scanAt = time.Now()
+ case "startup:scanned":
+ sc.doneAt = time.Now()
+ }
+ at(point, st, path)
+ })
+ defer TestHooks.Fault(nil)
+ var logs strings.Builder
+ restore := captureLogTo(&logs)
+ srv, 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)
}
- })
- 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())
+ if sc.scanAt.IsZero() || sc.doneAt.IsZero() {
+ t.Fatal("the start scan never ran")
+ }
+ if took := sc.doneAt.Sub(sc.scanAt); took > selfServeEvidenceTimeout+500*time.Millisecond {
+ t.Fatalf("the start scan took %v, past its bound %v", took, selfServeEvidenceTimeout)
+ }
+ sc.srv, sc.logs = srv, logs.String()
+ return sc
}
- 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)
+ failedClosed := func(t *testing.T, sc scan) {
+ t.Helper()
+ found, err := sc.srv.SelfServeStartEvidence()
+ if !found || !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("the start scan answered %v %v; want found with its deadline exceeded", found, err)
+ }
+ if !sc.srv.selfServeMode() || sc.srv.selfServeAdmits() {
+ t.Fatalf("mode %v admits %v; want recovery mode (fail closed)", sc.srv.selfServeMode(), sc.srv.selfServeAdmits())
+ }
+ if !strings.Contains(sc.logs, "recovery mode on") {
+ t.Fatalf("the abandoned journal read was not logged: %q", sc.logs)
+ }
}
+ t.Run("a SQLite write lock held on the file past the bound", func(t *testing.T) {
+ sc := boot(t, func(point string, _ *Store, path string) {
+ if point != "startup:scan" {
+ return
+ }
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ conn, err := db.Conn(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := conn.ExecContext(context.Background(), "BEGIN IMMEDIATE"); err != nil {
+ t.Fatalf("BEGIN IMMEDIATE: %v", err)
+ }
+ time.AfterFunc(hold, func() {
+ _, _ = conn.ExecContext(context.Background(), "ROLLBACK")
+ _ = conn.Close()
+ _ = db.Close()
+ })
+ })
+ if found, err := sc.srv.SelfServeStartEvidence(); found || err != nil {
+ t.Fatalf("with a writer holding the lock the scan answered %v %v; want the journal read as it is (WAL)", found, err)
+ }
+ })
+ t.Run("the store's one pooled connection held", func(t *testing.T) {
+ failedClosed(t, boot(t, func(point string, st *Store, _ string) {
+ if point != "startup:scan" {
+ return
+ }
+ tx, err := st.db.BeginTx(context.Background(), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ time.AfterFunc(hold, func() { _ = tx.Rollback() })
+ }))
+ })
+ t.Run("a read that blocks inside the driver", func(t *testing.T) {
+ failedClosed(t, boot(t, func(point string, _ *Store, _ string) {
+ if point == "evidence:journal" {
+ time.Sleep(hold) // no context reaches it
+ }
+ }))
+ })
}