amadan

carlos / platform Public

console: a refused takeover keeps its empty base; a late write is not published (#328, #329)

Review fix round 1 for the app policy door. An unverifiable object in a
new holding no longer marks the holding until a write records new
content, so a refused (412) PUT cannot leave the next write building on
the older recorded allow. A store that answers success after the
attempt's holding ended is the lease-lost refusal, the row left pending.
The door half of the lost-write-and-replay test is added.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c5632c461791eb0207261d0a9904edd29a8f1357 parent e131b5e
2 files changed, +125 −7
  • internal/console/apipolicy.go +22 −1
  • internal/console/apipolicy_test.go +103 −6
diff --git a/internal/console/apipolicy.go b/internal/console/apipolicy.go
index 553fd0bce..d09f922ab 100644
--- a/internal/console/apipolicy.go
+++ b/internal/console/apipolicy.go
@@ -221,6 +221,10 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
Wire: wire, SyncedETag: etag}
return nil
}
+ // durable is false when the base chosen is not represented in the
+ // table, so marking the holding would change what the next request
+ // builds on.
+ durable := true
notBuilding := func(why error) {
s.policyLogf("console: app policy %s/%s: not building on the stored object: %v", acct.Sqid, app, why)
}
@@ -235,8 +239,15 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
// A new holding, and the bucket holds something this console did not
// last publish: another console may have written it.
if b == nil {
+ // Nothing to adopt, so the takeover's empty base lives only in
+ // this decision. The holding is not marked: until a write
+ // records new content (writePolicy marks it then), every
+ // request of this holding decides again and starts empty,
+ // never on the older recorded content a refused write would
+ // otherwise expose.
notBuilding(gerr)
base = apppolicy.Policy{V: 1}
+ durable = false
break
}
if err := adopt(); err != nil {
@@ -287,7 +298,7 @@ func (s *Server) policyBaseFor(ctx context.Context, acct *Account, app string, h
base = apppolicy.Policy{V: 1}
}
}
- if commit {
+ if commit && durable {
s.markPolicyTouched(key, h.n)
}
return base, etag, max(recorded, bucketRev) + 1, row, cur, nil
@@ -525,6 +536,8 @@ func (s *Server) writePolicy(ctx context.Context, acct *Account, app, addr strin
s.policyLogf("console: app policy %s/%s: record revision %d: %v", acct.Sqid, app, next, err)
return policyReply{}, &doorRefusal{http.StatusServiceUnavailable, policyRecordFailedMsg}
}
+ // The content this holding builds on is now the table's.
+ s.markPolicyTouched(acct.Sqid+"/"+app, hold.n)
newETag, ref := s.publishPolicy(ctx, acct, app, hold, wire, next, etag)
if ref != nil {
return policyReply{}, ref
@@ -558,6 +571,14 @@ func (s *Server) publishPolicy(ctx context.Context, acct *Account, app string, h
return "", ref
}
newETag, err := apppolicy.PutWire(actx, s.cfg.Releases, acct.Sqid, app, wire, etag)
+ if err == nil && !hold.sole && actx.Err() != nil {
+ // The store answered after the attempt's holding check expired:
+ // a write that outlived its holding is never reported published.
+ // The row stays pending.
+ done()
+ s.policyLogf("console: app policy %s/%s: publish revision %d: the writer holding ended during the write", acct.Sqid, app, next)
+ return "", lost
+ }
if err == nil {
done()
return newETag, nil
diff --git a/internal/console/apipolicy_test.go b/internal/console/apipolicy_test.go
index 498dcdd80..df8725367 100644
--- a/internal/console/apipolicy_test.go
+++ b/internal/console/apipolicy_test.go
@@ -714,7 +714,7 @@ type appPolicyHookStore struct {
key string
mu sync.Mutex
puts int
- hook func(n int, body []byte, ifMatch string) (handled bool, etag string, err error)
+ hook func(ctx context.Context, n int, body []byte, ifMatch string) (handled bool, etag string, err error)
}
func (s *appPolicyHookStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
@@ -726,7 +726,7 @@ func (s *appPolicyHookStore) Put(ctx context.Context, key string, body []byte, i
n, hook := s.puts, s.hook
s.mu.Unlock()
if hook != nil {
- if handled, etag, err := hook(n, body, ifMatch); handled {
+ if handled, etag, err := hook(ctx, n, body, ifMatch); handled {
return etag, err
}
}
@@ -760,7 +760,7 @@ func TestPolicyPublishRechecksTheHoldingOnEveryAttempt(t *testing.T) {
t.Run("the lease fails during a conflicting attempt", func(t *testing.T) {
t.Parallel()
fx, f, hs := setup(t, "code-policy-attempt-err")
- hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ hs.hook = func(_ context.Context, n int, _ []byte, _ string) (bool, string, error) {
if n == 1 {
lastPolicyLease(t, f).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
return true, "", conflict
@@ -799,7 +799,7 @@ func TestPolicyPublishRechecksTheHoldingOnEveryAttempt(t *testing.T) {
t.Run("the safe deadline moves in during a conflicting attempt", func(t *testing.T) {
t.Parallel()
fx, f, hs := setup(t, "code-policy-attempt-moved")
- hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ hs.hook = func(_ context.Context, n int, _ []byte, _ string) (bool, string, error) {
if n == 1 {
lastPolicyLease(t, f).set(nil, time.Now().Add(8*time.Second), nil)
return true, "", conflict
@@ -818,7 +818,7 @@ func TestPolicyPublishRechecksTheHoldingOnEveryAttempt(t *testing.T) {
fx, _, hs := setup(t, "code-policy-attempt-other")
theirs := signPolicy(t, fx.consoleFixture, apppolicy.Policy{V: 1, Account: fx.sqid, App: "shop", Revision: 1,
SignedAt: time.Now().UTC(), DefaultSize: "large"})
- hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ hs.hook = func(_ context.Context, n int, _ []byte, _ string) (bool, string, error) {
if n == 1 {
if _, err := fx.rel.Put(context.Background(), apppolicy.Key(fx.sqid, "shop"), theirs, ""); err != nil {
t.Error(err)
@@ -837,10 +837,34 @@ func TestPolicyPublishRechecksTheHoldingOnEveryAttempt(t *testing.T) {
}
})
+ // Review fix round 1, finding 2: a store that answers success after the
+ // attempt's holding has ended is not a publication.
+ t.Run("a write answered after its holding ended is not published", func(t *testing.T) {
+ t.Parallel()
+ fx, f, hs := setup(t, "code-policy-attempt-late")
+ hs.hook = func(ctx context.Context, n int, body []byte, ifMatch string) (bool, string, error) {
+ l := lastPolicyLease(t, f)
+ l.mu.Lock()
+ close(l.lost)
+ l.mu.Unlock()
+ <-ctx.Done()
+ etag, err := fx.rel.Put(context.Background(), apppolicy.Key(fx.sqid, "shop"), body, ifMatch)
+ return true, etag, err
+ }
+ st, _, out := fx.put(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ wantLost(t, st, out)
+ if hs.putCount() != 1 {
+ t.Fatalf("%d Puts, want 1", hs.putCount())
+ }
+ if row := fx.row(t, "shop"); row.Revision != 1 || !row.Pending || row.SyncedETag != "" {
+ t.Fatalf("row %+v, want revision 1 still pending", row)
+ }
+ })
+
t.Run("control: a plain conflict is retried", func(t *testing.T) {
t.Parallel()
fx, _, hs := setup(t, "code-policy-attempt-control")
- hs.hook = func(n int, _ []byte, _ string) (bool, string, error) {
+ hs.hook = func(_ context.Context, n int, _ []byte, _ string) (bool, string, error) {
if n == 1 {
return true, "", conflict
}
@@ -1032,3 +1056,76 @@ func TestPolicyPutStartsEmptyOverAnotherIncarnationsPolicy(t *testing.T) {
t.Fatal("no app-policy event")
}
}
+
+// Review fix round 1, finding 1: in a new holding over an object that does
+// not verify, a refused write must not leave the holding marked, or the next
+// write would build on the older recorded content and bring its allow back.
+func TestPolicyRefusedTakeoverKeepsTheEmptyBase(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-takeover-refused")
+ f := fx.leased()
+ fx.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ lastPolicyLease(t, f).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ etag := putAtKey(t, fx.rel, fx.sqid, "shop", []byte(`{`))
+
+ if st, _, out := fx.put(t, "shop", `{"default_size":"small"}`, policyIfMatch("stale")); st != http.StatusPreconditionFailed {
+ t.Fatalf("PUT with a stale etag = %d %s, want 412", st, out)
+ }
+ if f.callCount() != 2 {
+ t.Fatalf("%d lease acquisitions, want 2 (the refused PUT took the new holding)", f.callCount())
+ }
+ if g := fx.get(t, "shop"); len(g.Policy.EgressAllow) != 0 || g.ETag != etag {
+ t.Fatalf("GET after the refused PUT = %+v etag %q, want no egress allow", g.Policy, g.ETag)
+ }
+ r := fx.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(etag))
+ stored, _, _ := fx.stored(t, "shop")
+ if len(r.Policy.EgressAllow) != 0 || len(stored.EgressAllow) != 0 || stored.DefaultSize != "small" || stored.Revision != 2 {
+ t.Fatalf("stored %+v after a size-only PUT, want revision 2 with no egress allow", stored)
+ }
+ // The write recorded the takeover's content: the holding builds on it.
+ r = fx.mustPut(t, "shop", `{"default_size":"large"}`, policyIfMatch(r.ETag))
+ if len(r.Policy.EgressAllow) != 0 || r.Policy.Revision != 3 {
+ t.Fatalf("the next PUT = %+v", r.Policy)
+ }
+}
+
+// Astra plan R1-1, the door's half (review fix round 1, finding 3). Task 10
+// adds the re-sign continuation: a pass between the replay and the GET
+// finishes the interrupted publication.
+func TestPolicyRevokeSurvivesALostWriteAndAReplay(t *testing.T) {
+ t.Parallel()
+ fx := newPolicyFx(t, "code-policy-lost-write")
+ r := fx.mustPut(t, "shop", `{"egress_allow":["10.0.5.20"]}`, policyCreate)
+ _, w1, _ := fx.stored(t, "shop")
+ hs := &appPolicyHookStore{Store: fx.rel, key: apppolicy.Key(fx.sqid, "shop")}
+ hs.hook = func(ctx context.Context, n int, body []byte, ifMatch string) (bool, string, error) {
+ if n == 1 {
+ if _, err := fx.rel.Put(ctx, apppolicy.Key(fx.sqid, "shop"), body, ifMatch); err != nil {
+ t.Error(err)
+ }
+ return true, "", errors.New("test: the reply was lost")
+ }
+ return false, "", nil
+ }
+ fx.srv.cfg.Releases = hs
+
+ st, _, out := fx.put(t, "shop", `{"egress_allow":[]}`, policyIfMatch(r.ETag))
+ if st != http.StatusBadGateway || errorOf(t, out) != policyUnpublishedMsg {
+ t.Fatalf("PUT whose write landed and errored = %d %s, want 502 %q", st, out, policyUnpublishedMsg)
+ }
+ row := fx.row(t, "shop")
+ recorded, err := apppolicy.OpenKey(fx.pub, row.Wire, fx.sqid, "shop")
+ if err != nil || row.Revision != 2 || recorded.Revision != 2 || len(recorded.EgressAllow) != 0 || !row.Pending {
+ t.Fatalf("row %+v (%+v, %v), want the revocation recorded at revision 2, pending", row, recorded, err)
+ }
+
+ etag := putAtKey(t, fx.rel, fx.sqid, "shop", w1)
+ g := fx.get(t, "shop")
+ if len(g.Policy.EgressAllow) != 0 || g.Policy.Revision != 2 || g.ETag != etag {
+ t.Fatalf("GET over the replayed allow = %+v etag %q, want the recorded revocation", g.Policy, g.ETag)
+ }
+ fx.mustPut(t, "shop", `{"default_size":"small"}`, policyIfMatch(etag))
+ if stored, _, _ := fx.stored(t, "shop"); stored.Revision != 3 || len(stored.EgressAllow) != 0 || stored.DefaultSize != "small" {
+ t.Fatalf("stored %+v, want revision 3 with no egress allow", stored)
+ }
+}