amadan

oficina / home Public

Make roster lease checks nonblocking during durable publication

Paul Campbell pushed by paul@keymail.dev b3bd3b7aeeba53557f6942ae692b298d724291de parent 3e874a0
5 files changed, +264 −3
  • roster/cache.go +8 −2
  • roster/go.mod +13 −0
  • roster/go.sum +31 −0
  • roster/lease_test.go +207 −0
  • roster/protocol.go +5 −1

Intent

full ledger: feat-roster-http

diff --git a/roster/cache.go b/roster/cache.go
index 1c645ea..e521bc4 100644
--- a/roster/cache.go
+++ b/roster/cache.go
@@ -68,6 +68,8 @@ type Decision struct {
// Lease is an observation, not a stored authorization grant. Call Valid at the
// protected operation boundary; stream users must recheck and re-arm their timer
// from a new decision after renewal, or close when renewal cannot authorize it.
+// Both accessors refuse immediately on cache contention: a caller may hold the
+// database writer that an in-flight durable publication needs to acquire.
type Lease struct {
cache *Cache
version uint64
@@ -79,7 +81,9 @@ func (l Lease) Valid() bool {
return false
}
c := l.cache
- c.mu.Lock()
+ if !c.mu.TryLock() {
+ return false
+ }
defer c.mu.Unlock()
_, n := c.observe()
return c.version == l.version && c.usable && n < l.deadline
@@ -89,7 +93,9 @@ func (l Lease) Remaining() time.Duration {
return 0
}
c := l.cache
- c.mu.Lock()
+ if !c.mu.TryLock() {
+ return 0
+ }
defer c.mu.Unlock()
_, n := c.observe()
if c.version != l.version || !c.usable || n >= l.deadline {
diff --git a/roster/go.mod b/roster/go.mod
index cdb9ca3..aa4ddb0 100644
--- a/roster/go.mod
+++ b/roster/go.mod
@@ -6,4 +6,17 @@ require (
amadan.net/oficina/home/handoff v0.0.0-20260908102137-481d2d57f489
amadan.net/rastrillo/rastrillo v0.26.1-0.20260908075707-29e4491631bd
github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2
+ modernc.org/sqlite v1.55.0
+)
+
+require (
+ github.com/dustin/go-humanize v1.0.1 // indirect
+ github.com/google/uuid v1.6.0 // indirect
+ github.com/mattn/go-isatty v0.0.20 // indirect
+ github.com/ncruces/go-strftime v1.0.0 // indirect
+ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+ golang.org/x/sys v0.46.0 // indirect
+ modernc.org/libc v1.74.1 // indirect
+ modernc.org/mathutil v1.7.1 // indirect
+ modernc.org/memory v1.11.0 // indirect
)
diff --git a/roster/go.sum b/roster/go.sum
index bfcd60e..4cab028 100644
--- a/roster/go.sum
+++ b/roster/go.sum
@@ -6,21 +6,52 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2 h1:iizUGZ9pEquQS5jTGkh4AqeeHCMbfbjeb0zMt0aEFzs=
github.com/go-json-experiment/json v0.0.0-20250725192818-e39067aee2d2/go.mod h1:TiCD2a1pcmjd7YnhGH0f/zKNcCD06B029pHhzV23c2M=
+github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
+github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
+golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ=
+golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0=
+golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
+golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q=
+golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA=
+modernc.org/cc/v4 v4.29.0 h1:CXgwL8cvxmyzBQZzbSl/6xFtMCryb6u8IOqDci39cgc=
+modernc.org/cc/v4 v4.29.0/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI=
+modernc.org/ccgo/v4 v4.34.6 h1:sBgfIwyN0TQ9C5hwIeuqyeAKyMWnbvj2fvpF4L11uzU=
+modernc.org/ccgo/v4 v4.34.6/go.mod h1:SZ8YcN9NG7XVsQYdm6jYBvi8PQP1qi+kqB6OhjqI3Fk=
+modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM=
+modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU=
+modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI=
+modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
+modernc.org/gc/v3 v3.1.4 h1:2g65LGVSmFQrXeITAw97x7hCRvZFcyE1uDP+7Vng7JI=
+modernc.org/gc/v3 v3.1.4/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY=
+modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
+modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
modernc.org/libc v1.74.1 h1:bdR4VTKFMC4966QSNZ05XLGI/VwzVa2kTUX51Dm0riQ=
modernc.org/libc v1.74.1/go.mod h1:uH4t5bOx3G3g9Xcmj10YKlTcVISlRDwv8VoQJG9n8Os=
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
+modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg=
+modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
+modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
+modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
modernc.org/sqlite v1.55.0 h1:hIFh0MCH0rGinQ/4KYb5/UbCkRkb+UP+OkLCVWa5MTM=
modernc.org/sqlite v1.55.0/go.mod h1:4ntCLuNmnH8+GNqjka1wNg7KJd5/Hi5FYp8K+XQ7GZw=
+modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
+modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
+modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
+modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
diff --git a/roster/lease_test.go b/roster/lease_test.go
new file mode 100644
index 0000000..ff742b4
--- /dev/null
+++ b/roster/lease_test.go
@@ -0,0 +1,207 @@
+package roster
+
+import (
+ "context"
+ "database/sql"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ _ "modernc.org/sqlite"
+)
+
+type sqliteEvidence struct {
+ db *sql.DB
+ mu sync.Mutex
+ entering chan struct{}
+}
+
+func (s *sqliteEvidence) Load(ctx context.Context) ([]byte, error) {
+ var b []byte
+ e := s.db.QueryRowContext(ctx, "SELECT value FROM evidence WHERE id=1").Scan(&b)
+ if e == sql.ErrNoRows {
+ return nil, ErrNoEvidence
+ }
+ return b, e
+}
+func (s *sqliteEvidence) Save(ctx context.Context, b []byte) error {
+ s.mu.Lock()
+ entered := s.entering
+ s.entering = nil
+ s.mu.Unlock()
+ if entered != nil {
+ close(entered)
+ }
+ _, e := s.db.ExecContext(ctx, "INSERT INTO evidence(id,value) VALUES(1,?) ON CONFLICT(id) DO UPDATE SET value=excluded.value", b)
+ return e
+}
+
+func TestLeaseDoesNotWaitForSQLiteWriterPublication(t *testing.T) {
+ f := setup(t)
+ db, e := sql.Open("sqlite", filepath.Join(t.TempDir(), "roster.db"))
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer db.Close()
+ db.SetMaxOpenConns(1)
+ if _, e = db.Exec("CREATE TABLE evidence(id INTEGER PRIMARY KEY,value BLOB NOT NULL)"); e != nil {
+ t.Fatal(e)
+ }
+ disk := &sqliteEvidence{db: db}
+ f.cfg.Durable = disk
+ c := f.cache()
+ d := f.check(c, 1)
+ state(t, d, Member)
+ tx, e := db.BeginTx(context.Background(), nil)
+ if e != nil {
+ t.Fatal(e)
+ }
+ defer tx.Rollback()
+ if _, e = tx.Exec("UPDATE evidence SET value=value WHERE id=1"); e != nil {
+ t.Fatal(e)
+ }
+ before := db.Stats().WaitCount
+ entered := make(chan struct{})
+ disk.mu.Lock()
+ disk.entering = entered
+ disk.mu.Unlock()
+ f.clock.step(30 * time.Second)
+ refreshed := make(chan error, 1)
+ go func() { refreshed <- c.Refresh(context.Background()) }()
+ <-entered
+ // WaitCount proves Save is waiting on the real one-connection writer pool,
+ // not merely scheduled to enter a fake callback. Publication owns Cache.mu.
+ deadline := time.Now().Add(time.Second)
+ for db.Stats().WaitCount == before && time.Now().Before(deadline) {
+ time.Sleep(time.Millisecond)
+ }
+ if db.Stats().WaitCount == before {
+ _ = tx.Rollback()
+ t.Fatal("publication never waited for writer")
+ }
+ valid := make(chan bool, 1)
+ remaining := make(chan time.Duration, 1)
+ go func() { valid <- d.Lease.Valid() }()
+ go func() { remaining <- d.Lease.Remaining() }()
+ var v bool
+ var left time.Duration
+ timedOut := false
+ select {
+ case v = <-valid:
+ case <-time.After(250 * time.Millisecond):
+ timedOut = true
+ }
+ select {
+ case left = <-remaining:
+ case <-time.After(250 * time.Millisecond):
+ timedOut = true
+ }
+ // Always release the writer before reporting a mutation failure, so neither
+ // the refresh goroutine nor the blocked accessor can leak into another test.
+ _ = tx.Rollback()
+ if e = <-refreshed; e != nil {
+ t.Fatal(e)
+ }
+ if timedOut {
+ t.Fatal("lease accessor blocked behind SQLite writer publication")
+ }
+ if v || left != 0 {
+ t.Fatal("contended lease granted standing")
+ }
+ next := f.check(c, 1)
+ state(t, next, Member)
+ if !next.Lease.Valid() {
+ t.Fatal("retry could not use published lease")
+ }
+ if d.Lease.Valid() {
+ t.Fatal("old lease survived publication version")
+ }
+}
+
+func TestLeaseContentionRefusesWithoutChangingAgeOrStanding(t *testing.T) {
+ f := setup(t)
+ c := f.cache()
+ d := f.check(c, 1)
+ f.clock.step(10 * time.Second)
+ before := d.Lease.Remaining()
+ c.mu.Lock()
+ valid := make(chan bool, 1)
+ left := make(chan time.Duration, 1)
+ go func() { valid <- d.Lease.Valid() }()
+ go func() { left <- d.Lease.Remaining() }()
+ timedOut := false
+ var v bool
+ var n time.Duration
+ select {
+ case v = <-valid:
+ case <-time.After(250 * time.Millisecond):
+ timedOut = true
+ }
+ select {
+ case n = <-left:
+ case <-time.After(250 * time.Millisecond):
+ timedOut = true
+ }
+ c.mu.Unlock()
+ if timedOut {
+ t.Fatal("contended accessor waited")
+ }
+ if v || n != 0 {
+ t.Fatal("contention granted standing")
+ }
+ if !d.Lease.Valid() || d.Lease.Remaining() != before {
+ t.Fatal("contention changed the snapshot or deadline")
+ }
+ f.clock.step(50 * time.Second)
+ if d.Lease.Valid() || d.Lease.Remaining() != 0 {
+ t.Fatal("contention extended age")
+ }
+}
+
+func TestLeaseContentionDoesNotSkipLaterClockFence(t *testing.T) {
+ f := setup(t)
+ c := f.cache()
+ d := f.check(c, 1)
+ f.clock.wallStep(-ClockTolerance - time.Nanosecond)
+ c.mu.Lock()
+ done := make(chan bool, 1)
+ go func() { done <- d.Lease.Valid() }()
+ select {
+ case v := <-done:
+ if v {
+ c.mu.Unlock()
+ t.Fatal("contention granted")
+ }
+ case <-time.After(250 * time.Millisecond):
+ c.mu.Unlock()
+ t.Fatal("contention blocked")
+ }
+ c.mu.Unlock()
+ if d.Lease.Valid() || d.Lease.Remaining() != 0 {
+ t.Fatal("successful lock skipped backward-clock fence")
+ }
+}
+
+func TestUnknownSigningKeyRefused(t *testing.T) {
+ f := setup(t)
+ raw, e := SignProvisioning(f.op, "unprovisioned", f.record)
+ if e != nil {
+ t.Fatal(e)
+ }
+ if _, e = VerifyProvisioning(raw, []Key{{"operator", f.record.Keys[0].Public}}, f.record.Bindings); e != ErrRefused {
+ t.Fatalf("unknown operator key result %v", e)
+ }
+ keys, _ := publicKeys(f.record.Keys)
+ // A valid signature under a real key still refuses when its identifier was
+ // not provisioned. The map guard is explicit rather than verifier-dependent.
+ b, e := encode(f.key, "unprovisioned", PageContext, struct {
+ Version int `json:"version"`
+ }{1})
+ if e != nil {
+ t.Fatal(e)
+ }
+ if _, e = signedPayload(b, keys, PageContext); e != ErrRefused {
+ t.Fatalf("unknown roster key result %v", e)
+ }
+}
diff --git a/roster/protocol.go b/roster/protocol.go
index 558fb04..bcdb434 100644
--- a/roster/protocol.go
+++ b/roster/protocol.go
@@ -144,10 +144,14 @@ func signedPayload(data []byte, keys map[string][]byte, purpose string) ([]byte,
if strict(data, &e) != nil || e.Version != 1 || !atom(e.KeyID, 64) {
return nil, ErrRefused
}
+ key, known := keys[e.KeyID]
+ if !known {
+ return nil, ErrRefused
+ }
p, ok := canonical(e.Payload, -1)
sig, ok2 := canonical(e.Signature, 64)
// The signature covers decoded UTF-8 payload bytes, never base64url text.
- if !ok || !ok2 || !family.Verify(keys[e.KeyID], purpose, p, sig) {
+ if !ok || !ok2 || !family.Verify(key, purpose, p, sig) {
return nil, ErrRefused
}
return p, nil