amadan

carlos / platform Public

console: a listed app policy stays a re-sign candidate until a pass processes it (#328, #329)

Review fix round 1. The re-sign pass keeps every listed policy with no
row in policyDiscovered until one pass processes it, so a failed read or
adoption is retried on the next pass without another listing. The
worker test helper registers its cancel before it waits, and the
two-console witnesses capture and assert the lease's log lines.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ad52b088e494c800ad925cd38cb0d268a2355a36 parent 3e86e9d
4 files changed, +193 −27
  • internal/console/apipolicy.go +55 −19
  • internal/console/apipolicy_resign_test.go +80 −0
  • internal/console/policylease_twoconsoles_test.go +48 −1
  • internal/console/server.go +10 −7
diff --git a/internal/console/apipolicy.go b/internal/console/apipolicy.go
index ec9fb05ef..9eb2c6f6d 100644
--- a/internal/console/apipolicy.go
+++ b/internal/console/apipolicy.go
@@ -652,10 +652,13 @@ const policyResignAction = "app-policy-resign"
// holding is lost: the pass stops there.
var errPolicyResignLost = errors.New("console: app policies: the writer lease was lost")
-// policyCandidate is one app the re-sign pass reads.
+// policyCandidate is one app the re-sign pass reads. discovered is a
+// listed key with no row: it stays in policyDiscovered until a pass
+// processes it.
type policyCandidate struct {
- acct *Account
- app string
+ acct *Account
+ app string
+ discovered bool
}
// resignAppPolicies re-signs every live app policy with a fresh SignedAt, as
@@ -685,6 +688,9 @@ type policyCandidate struct {
// a holding, and on every pass while it holds none (another console may hold
// the lease and have written apps this table does not know), so a steady
// state costs one Get per live policy per pass and no listing (spec §3a).
+// A listed policy with no row is kept as a candidate across passes until one
+// processes it: a failed read or adoption is retried by the next pass, never
+// by another listing.
//
// Nothing is signed without the writer lease (ruling 10): with candidates
// the pass takes one holding, and every app's signing and publication is
@@ -715,12 +721,20 @@ func (s *Server) resignAppPolicies(ctx context.Context) {
}
if err != nil {
s.policyLogf("console: app policy %s/%s: re-sign: %v", c.acct.Sqid, c.app, err)
+ continue
+ }
+ if c.discovered {
+ // Processed: adopted into the table, found absent, or left
+ // alone as unverifiable. The row, or the next listing, carries
+ // it from here.
+ s.forgetDiscoveredPolicy(c.acct.Sqid, c.app)
}
}
}
// policyResignCandidates is the pass's apps, sorted: every live or pending
-// row, and, on a pass that lists, every listed key that has no row.
+// row, and every listed key that has no row and no pass has processed yet
+// (policyDiscovered), whether this pass listed or an earlier one did.
func (s *Server) policyResignCandidates(ctx context.Context) ([]policyCandidate, error) {
rows, err := s.store.AppPolicyRowsContext(ctx)
if err != nil {
@@ -730,7 +744,7 @@ func (s *Server) policyResignCandidates(ctx context.Context) ([]policyCandidate,
var out []policyCandidate
for _, r := range rows {
seen[r.Sqid+"/"+r.App] = true
- out = append(out, policyCandidate{&Account{ID: r.AccountID, Sqid: r.Sqid}, r.App})
+ out = append(out, policyCandidate{acct: &Account{ID: r.AccountID, Sqid: r.Sqid}, app: r.App})
}
n := s.policyReadHolding().n
@@ -744,30 +758,45 @@ func (s *Server) policyResignCandidates(ctx context.Context) ([]policyCandidate,
} else {
s.policyStateMu.Lock()
s.policyListedN = n
- s.policyStateMu.Unlock()
for key := range present {
sqid, app, ok := policyKeyApp(key)
if !ok {
s.policyLogf("console: app policies: skipping %s: not an app policy key", key)
continue
}
- if seen[sqid+"/"+app] {
- continue
- }
- seen[sqid+"/"+app] = true
- acct, err := s.store.AccountBySqid(sqid)
- if err != nil {
- s.policyLogf("console: app policy %s/%s: account: %v", sqid, app, err)
- continue
- }
- if acct == nil {
- s.policyLogf("console: app policy %s/%s: no such account; skipped (the purge sweep owns it)", sqid, app)
- continue
+ if !seen[sqid+"/"+app] {
+ if s.policyDiscovered == nil {
+ s.policyDiscovered = map[[2]string]bool{}
+ }
+ s.policyDiscovered[[2]string{sqid, app}] = true
}
- out = append(out, policyCandidate{acct, app})
}
+ s.policyStateMu.Unlock()
}
}
+ // Every listed key with no row, from this listing or an earlier one not
+ // yet processed.
+ s.policyStateMu.Lock()
+ discovered := slices.Collect(maps.Keys(s.policyDiscovered))
+ s.policyStateMu.Unlock()
+ for _, k := range discovered {
+ sqid, app := k[0], k[1]
+ if seen[sqid+"/"+app] {
+ s.forgetDiscoveredPolicy(sqid, app) // a row carries it now
+ continue
+ }
+ acct, err := s.store.AccountBySqid(sqid)
+ if err != nil {
+ s.policyLogf("console: app policy %s/%s: account: %v", sqid, app, err)
+ continue
+ }
+ if acct == nil {
+ s.policyLogf("console: app policy %s/%s: no such account; skipped (the purge sweep owns it)", sqid, app)
+ s.forgetDiscoveredPolicy(sqid, app)
+ continue
+ }
+ out = append(out, policyCandidate{acct: acct, app: app, discovered: true})
+ }
sort.Slice(out, func(i, j int) bool {
if out[i].acct.Sqid != out[j].acct.Sqid {
return out[i].acct.Sqid < out[j].acct.Sqid
@@ -777,6 +806,13 @@ func (s *Server) policyResignCandidates(ctx context.Context) ([]policyCandidate,
return out, nil
}
+// forgetDiscoveredPolicy drops a listed key from policyDiscovered.
+func (s *Server) forgetDiscoveredPolicy(sqid, app string) {
+ s.policyStateMu.Lock()
+ delete(s.policyDiscovered, [2]string{sqid, app})
+ s.policyStateMu.Unlock()
+}
+
// policyKeyApp is the account and app a listed key is the policy of: ok only
// when the key is exactly apppolicy.Key(sqid, app) for a valid account and
// app.
diff --git a/internal/console/apipolicy_resign_test.go b/internal/console/apipolicy_resign_test.go
index 262a76218..a6a13eae9 100644
--- a/internal/console/apipolicy_resign_test.go
+++ b/internal/console/apipolicy_resign_test.go
@@ -71,6 +71,9 @@ type policyCountingStore struct {
gets map[string]int
puts map[string]int
afterPut func(key string)
+ // failGet, when set, answers a policy Get with its error instead of
+ // the store when it returns one.
+ failGet func(key string) error
}
func newPolicyCountingStore(s release.Store) *policyCountingStore {
@@ -103,7 +106,13 @@ func (c *policyCountingStore) Get(ctx context.Context, key string) ([]byte, stri
if strings.HasPrefix(key, apppolicy.Prefix) {
c.mu.Lock()
c.gets[key]++
+ fail := c.failGet
c.mu.Unlock()
+ if fail != nil {
+ if err := fail(key); err != nil {
+ return nil, "", err
+ }
+ }
}
return c.Store.Get(ctx, key)
}
@@ -166,6 +175,15 @@ func runDoorKeyResignOnce(t *testing.T, srv *Server, done func() bool) {
t.Helper()
ctx, cancel := context.WithCancel(context.Background())
returned := make(chan struct{})
+ // Registered before the wait, so a failed wait still stops the worker.
+ t.Cleanup(func() {
+ cancel()
+ select {
+ case <-returned:
+ case <-time.After(10 * time.Second):
+ t.Error("StartDoorKeyResign did not return after its context was cancelled")
+ }
+ })
go func() {
defer close(returned)
srv.StartDoorKeyResign(ctx)
@@ -483,6 +501,68 @@ WHEN NEW.synced_etag != '' BEGIN SELECT RAISE(ABORT, 'test: the sync mark fails'
}
}
+// Review fix round 1, finding 1: a listed policy with no row whose first
+// read fails is still a candidate on the next pass, with no second listing.
+func TestResignRetriesADiscoveredPolicyWithoutListingAgain(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-resign-discovered")
+ fx.leased()
+ cs := newPolicyCountingStore(fx.rel)
+ fx.srv.cfg.Releases = cs
+ fx.mustPut(t, "shop", `{"default_size":"small"}`, policyCreate) // the holding
+ // A policy this console has no row for, as another console's.
+ signed := time.Now().UTC().Add(-50 * time.Hour).Truncate(time.Second)
+ putAtKey(t, fx.rel, fx.sqid, "shop-b", signPolicy(t, fx.consoleFixture, apppolicy.Policy{V: 1, Account: fx.sqid,
+ App: "shop-b", Revision: 3, SignedAt: signed, EgressAllow: []string{"10.0.5.20"}}))
+ key := apppolicy.Key(fx.sqid, "shop-b")
+ var failed bool
+ cs.mu.Lock()
+ cs.failGet = func(k string) error {
+ if k == key && !failed {
+ failed = true
+ return errors.New("test: the read failed")
+ }
+ return nil
+ }
+ cs.mu.Unlock()
+ cs.reset()
+ ctx := context.Background()
+
+ fx.srv.resignAppPolicies(ctx)
+ if !failed {
+ t.Fatal("the pass did not read the listed policy")
+ }
+ if row := fx.row(t, "shop-b"); row.Revision != 0 {
+ t.Fatalf("row %+v after a failed read", row)
+ }
+ fx.logs.has(t, "console: app policy bab/shop-b: re-sign: ")
+
+ fx.srv.resignAppPolicies(ctx)
+ got, err := apppolicy.VerifyKey(fx.pub, fx.bytesAt(t, "shop-b"), time.Now(), fx.sqid, "shop-b")
+ if err != nil || got.Revision != 3 || !got.SignedAt.After(signed) || !slices.Equal(got.EgressAllow, []string{"10.0.5.20"}) {
+ t.Fatalf("after the next pass %+v %v, want revision 3 re-signed", got, err)
+ }
+ _, _, etag := fx.stored(t, "shop-b")
+ if row := fx.row(t, "shop-b"); row.Revision != 3 || row.SyncedETag != etag || row.Pending {
+ t.Fatalf("row %+v, want revision 3 synced", row)
+ }
+ if lists, _, _, _ := cs.counts(); lists != 1 {
+ t.Fatalf("%d listings over the two passes, want 1", lists)
+ }
+
+ // Processed: the row carries it, and nothing is listed again.
+ fx.srv.resignAppPolicies(ctx)
+ if lists, _, _, _ := cs.counts(); lists != 1 {
+ t.Fatalf("%d listings over three passes, want 1", lists)
+ }
+ fx.srv.policyStateMu.Lock()
+ left := len(fx.srv.policyDiscovered)
+ fx.srv.policyStateMu.Unlock()
+ if left != 0 {
+ t.Fatalf("%d discovered policies still held after they were processed", left)
+ }
+}
+
func TestResignWithNoPoliciesTakesNoLease(t *testing.T) {
t.Parallel()
fx := newPolicyFx(t, "code-resign-none")
diff --git a/internal/console/policylease_twoconsoles_test.go b/internal/console/policylease_twoconsoles_test.go
index 7569cf8da..49d0473f1 100644
--- a/internal/console/policylease_twoconsoles_test.go
+++ b/internal/console/policylease_twoconsoles_test.go
@@ -6,9 +6,11 @@ import (
"encoding/json"
"errors"
"fmt"
+ "log"
"net/http"
"net/http/httptest"
"slices"
+ "strings"
"sync/atomic"
"testing"
"time"
@@ -36,10 +38,35 @@ const policyLeaseLockKey = appPolicyLeasePrefix + "/lock.json"
// twoConsoles is the witnesses' world: one versioned bucket "deploy" behind
// an S3 server, and consoles A and B over it with the same release key.
+// Each console's policy lines go to its policyFx's logs (the server's seam);
+// internal/lease and the console's start log through the process logger,
+// which procLog captures for the test (these tests are not parallel).
type twoConsoles struct {
bucket *leasetest.VersionedBucket
failLock atomic.Bool
a, b *policyFx
+ procLog *policyLogRecorder
+}
+
+// policyLogWriter is a log.Logger output that records each line.
+type policyLogWriter struct{ r *policyLogRecorder }
+
+func (w policyLogWriter) Write(p []byte) (int, error) {
+ w.r.logf("%s", strings.TrimSuffix(string(p), "\n"))
+ return len(p), nil
+}
+
+// count is how many recorded lines contain sub.
+func (r *policyLogRecorder) count(sub string) int {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ n := 0
+ for _, l := range r.lines {
+ if strings.Contains(l, sub) {
+ n++
+ }
+ }
+ return n
}
func newTwoConsoles(t *testing.T) *twoConsoles {
@@ -50,7 +77,14 @@ func newTwoConsoles(t *testing.T) *twoConsoles {
if policyWitnessTTL-policyWitnessTTL/3-lease.RenewMargin < policyAttemptWindow {
t.Fatalf("policyWitnessTTL %s leaves a live holding less than policyAttemptWindow %s", policyWitnessTTL, policyAttemptWindow)
}
- w := &twoConsoles{bucket: leasetest.NewVersionedBucket()}
+ w := &twoConsoles{bucket: leasetest.NewVersionedBucket(), procLog: &policyLogRecorder{}}
+ prevOut, prevFlags := log.Writer(), log.Flags()
+ log.SetOutput(policyLogWriter{w.procLog})
+ log.SetFlags(0)
+ t.Cleanup(func() {
+ log.SetOutput(prevOut)
+ log.SetFlags(prevFlags)
+ })
w.bucket.AddBucket("deploy")
// Set once, before the server serves: the server reads it under the
// bucket's own lock, and the test flips failLock.
@@ -159,6 +193,7 @@ func restartPolicyWriter(s *Server) {
s.policyStateMu.Lock()
s.policyStopping = false
s.policyListedN = 0
+ s.policyDiscovered = nil
s.policyStateMu.Unlock()
s.policySyncedMu.Lock()
s.policySynced = nil
@@ -242,6 +277,12 @@ func TestPolicyWriterLeaseIsOneConsolePerBucket(t *testing.T) {
if !bytes.Equal(A.bucketBytes(t), wire) {
t.Fatal("A's refused re-sign changed the object")
}
+ A.logs.has(t, "console: policy writer lease lost: ")
+ A.logs.has(t, "console: app policies: not re-signing; another console (")
+ // B, A, then B again after A's lapse.
+ if n := w.procLog.count("lease: " + appPolicyLeasePrefix + ": acquired in"); n != 3 {
+ t.Errorf("%d lease acquisitions logged, want 3: %q", n, w.procLog.lines)
+ }
}
// Astra plan R3-1: a writer paused inside a write must not overwrite the
@@ -342,4 +383,10 @@ func TestALapsedWriterCannotOverwriteTheNextWritersRevoke(t *testing.T) {
got.Revision, got.Pending, rowB.Revision, rowB.Pending)
}
B.logs.has(t, "console: app policies: stopped re-signing at bab/shop; the writer lease was lost")
+ A.logs.has(t, "console: policy writer lease lost: ")
+ B.logs.has(t, "console: policy writer lease lost: ")
+ // A, B, then A again after B's lapse.
+ if n := w.procLog.count("lease: " + appPolicyLeasePrefix + ": acquired in"); n != 3 {
+ t.Errorf("%d lease acquisitions logged, want 3: %q", n, w.procLog.lines)
+ }
}
diff --git a/internal/console/server.go b/internal/console/server.go
index b0369d2d4..6ce9dc44b 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -1005,13 +1005,16 @@ type Server struct {
// production), called right after policyBase returns in a write or a
// re-sign, so a test can pause a writer that has read its base.
// policyListedN, under policyStateMu, is the holding the re-sign pass
- // last listed control/app-policy/ in (0: not holding, never counted).
- policyLocksMu sync.Mutex
- policyLocks map[string]*sync.Mutex
- policySyncedMu sync.Mutex
- policySynced map[string]uint64
- policyAfterBase func(sqid, app string)
- policyListedN uint64
+ // last listed control/app-policy/ in (0: not holding, never counted);
+ // policyDiscovered, also under policyStateMu, is every listed
+ // {sqid, app} with no row that no pass has processed yet.
+ policyLocksMu sync.Mutex
+ policyLocks map[string]*sync.Mutex
+ policySyncedMu sync.Mutex
+ policySynced map[string]uint64
+ policyAfterBase func(sqid, app string)
+ policyListedN uint64
+ policyDiscovered map[[2]string]bool
// sendApp is how the alert watcher (alertwatch.go) delivers keymail:
// km.SendApp in production, a capture in tests. A field rather than a
// direct call so an alertPass assertion needs no keymail server.