amadan

carlos / platform Public

release: concrete stores bump the change index after each mutation (#336)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude 559295c78362a0be9777b88f4c2fe05703cc86ba parent 2113b70
4 files changed, +387 −6
  • internal/release/changeindex.go +29 −0
  • internal/release/changeindex_store_test.go +245 −0
  • internal/release/conditional_delete.go +17 −1
  • internal/release/store.go +96 −5
diff --git a/internal/release/changeindex.go b/internal/release/changeindex.go
index ccd4cfa2a..82973d523 100644
--- a/internal/release/changeindex.go
+++ b/internal/release/changeindex.go
@@ -12,6 +12,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "log"
"math/rand/v2"
"os"
"strings"
@@ -149,3 +150,31 @@ func bumpChangeIndex(ctx context.Context,
}
return fmt.Errorf("bump %s: lost CAS race %d times: %w", ChangeIndexKey, changeIndexAttempts, lastErr)
}
+
+// changeIndexBumpBudget bounds one post-mutation bump, retries included. A
+// var so tests can shrink it.
+var changeIndexBumpBudget = 2 * time.Second
+
+// bumpAfterMutation is the hook both concrete stores call after a successful
+// mutation of key. It does nothing unless the store enabled the index and key
+// falls in a change scope. It runs under a context detached from the caller's
+// cancellation (a request that returns right after its write must not drop the
+// bump) but bounded by changeIndexBumpBudget, and it takes the stores' raw,
+// unhooked get/put so writing the index never bumps again. A failure is logged
+// and swallowed: the write already happened.
+func bumpAfterMutation(ctx context.Context, enabled bool, key string,
+ get func(context.Context, string) ([]byte, string, error),
+ put func(context.Context, string, []byte, string) (string, error)) {
+ if !enabled {
+ return
+ }
+ scope, ok := ChangeScope(key)
+ if !ok {
+ return
+ }
+ bctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), changeIndexBumpBudget)
+ defer cancel()
+ if err := bumpChangeIndex(bctx, get, put, scope); err != nil {
+ log.Printf("release: change index bump %s: %v", scope, err)
+ }
+}
diff --git a/internal/release/changeindex_store_test.go b/internal/release/changeindex_store_test.go
new file mode 100644
index 000000000..d5fa2f342
--- /dev/null
+++ b/internal/release/changeindex_store_test.go
@@ -0,0 +1,245 @@
+package release
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
+ "testing"
+)
+
+const bumpTestKey = "apps/bab/shop/instances/h.json"
+const bumpTestScope = "apps/bab/shop"
+
+func indexedDirStore(t *testing.T) *DirStore {
+ t.Helper()
+ d := NewDirStore(t.TempDir())
+ d.EnableChangeIndex()
+ return d
+}
+
+func scopeGen(t *testing.T, s Store, scope string) uint64 {
+ t.Helper()
+ idx, _, err := ReadChangeIndex(context.Background(), s)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return idx.Entries[scope]
+}
+
+func TestStoreBumpsAfterEveryMutation(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := indexedDirStore(t)
+ want := uint64(0)
+ step := func(name string, err error) {
+ t.Helper()
+ if err != nil {
+ t.Fatalf("%s: %v", name, err)
+ }
+ want++
+ if got := scopeGen(t, d, bumpTestScope); got != want {
+ t.Fatalf("after %s: gen = %d, want %d", name, got, want)
+ }
+ }
+ etag, err := d.Put(ctx, bumpTestKey, []byte("one"), "")
+ step("Put", err)
+ etag, err = d.PutFrom(ctx, bumpTestKey, bytes.NewReader([]byte("two")), 3, etag)
+ step("PutFrom", err)
+ step("DeleteIfMatch", DeleteIfMatch(ctx, d, bumpTestKey, etag))
+ _, err = d.Put(ctx, bumpTestKey, []byte("three"), "")
+ step("Put again", err)
+ step("Delete", d.Delete(ctx, bumpTestKey))
+}
+
+func TestStoreDoesNotBumpFailedMutation(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := indexedDirStore(t)
+ if _, err := d.Put(ctx, bumpTestKey, []byte("one"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := d.Put(ctx, bumpTestKey, []byte("two"), "stale"); !errors.Is(err, ErrCASConflict) {
+ t.Fatalf("stale Put = %v, want ErrCASConflict", err)
+ }
+ if _, err := d.PutFrom(ctx, bumpTestKey, bytes.NewReader([]byte("two")), 3, "stale"); !errors.Is(err, ErrCASConflict) {
+ t.Fatalf("stale PutFrom = %v, want ErrCASConflict", err)
+ }
+ if err := DeleteIfMatch(ctx, d, bumpTestKey, "stale"); !errors.Is(err, ErrCASConflict) {
+ t.Fatalf("stale DeleteIfMatch = %v, want ErrCASConflict", err)
+ }
+ if got := scopeGen(t, d, bumpTestScope); got != 1 {
+ t.Fatalf("gen = %d after one success and three refusals, want 1", got)
+ }
+}
+
+func TestStoreDoesNotBumpUnscopedKeys(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := indexedDirStore(t)
+ for _, k := range []string{"apps/bab/shop/logs/box/x", "control/serial", "status/boxes/a.json", "releases/x"} {
+ if _, err := d.Put(ctx, k, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := d.Delete(ctx, k); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if _, _, err := d.Get(ctx, ChangeIndexKey); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("index object after unscoped writes: err = %v, want not-exist", err)
+ }
+ // Writing the index itself never recurses or bumps.
+ if _, err := d.Put(ctx, ChangeIndexKey, []byte(`{"schema":1,"entries":{}}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ if got := scopeGen(t, d, bumpTestScope); got != 0 {
+ t.Fatalf("gen = %d after writing the index key, want 0", got)
+ }
+}
+
+func TestStoreBumpFailureDoesNotFailWrite(t *testing.T) {
+ ctx := context.Background()
+ d := indexedDirStore(t)
+ // A directory where the index object belongs: reading it fails with
+ // something other than not-exist, so every bump attempt errors.
+ if err := os.MkdirAll(filepath.Join(d.root, filepath.FromSlash(ChangeIndexKey)), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ var logs bytes.Buffer
+ logTo(t, &logs)
+ if _, err := d.Put(ctx, bumpTestKey, []byte("one"), ""); err != nil {
+ t.Fatalf("Put failed because the bump did: %v", err)
+ }
+ body, _, err := d.Get(ctx, bumpTestKey)
+ if err != nil || string(body) != "one" {
+ t.Fatalf("record = %q, %v; want written", body, err)
+ }
+ if !strings.Contains(logs.String(), "release: change index bump "+bumpTestScope+":") {
+ t.Fatalf("bump failure was not logged: %q", logs.String())
+ }
+}
+
+func TestStoreBumpDetachedFromCallerCancel(t *testing.T) {
+ t.Parallel()
+ d := indexedDirStore(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ var seenCancelled atomic.Bool
+ get := func(c context.Context, k string) ([]byte, string, error) {
+ if c.Err() != nil {
+ seenCancelled.Store(true)
+ }
+ return d.Get(c, k)
+ }
+ put := func(c context.Context, k string, b []byte, m string) (string, error) {
+ if c.Err() != nil {
+ seenCancelled.Store(true)
+ }
+ return d.putRaw(c, k, b, m)
+ }
+ // The caller's request ends the moment its write returns.
+ cancel()
+ bumpAfterMutation(ctx, true, bumpTestKey, get, put)
+ if seenCancelled.Load() {
+ t.Fatal("the bump ran under the caller's cancelled context")
+ }
+ if got := scopeGen(t, d, bumpTestScope); got != 1 {
+ t.Fatalf("gen = %d after a bump with a cancelled caller, want 1", got)
+ }
+ // And through the store: a cancelled context still writes (DirStore does
+ // not look at it) and still bumps.
+ cctx, ccancel := context.WithCancel(context.Background())
+ ccancel()
+ if _, err := d.Put(cctx, bumpTestKey, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if got := scopeGen(t, d, bumpTestScope); got != 2 {
+ t.Fatalf("gen = %d, want 2", got)
+ }
+}
+
+func TestStoreWithoutEnableNeverBumps(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := NewDirStore(t.TempDir())
+ etag, err := d.Put(ctx, bumpTestKey, []byte("one"), "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := d.PutFrom(ctx, bumpTestKey, bytes.NewReader([]byte("two")), 3, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := d.Delete(ctx, bumpTestKey); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := d.Get(ctx, ChangeIndexKey); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("index object on a store that never enabled it: err = %v", err)
+ }
+}
+
+func TestStoreCapabilitiesPreserved(t *testing.T) {
+ t.Parallel()
+ d := NewDirStore(t.TempDir())
+ d.EnableChangeIndex()
+ var s Store = d
+ _, ok1 := s.(ConditionalDeleter)
+ _, ok2 := s.(PutStreamer)
+ _, ok3 := s.(Opener)
+ _, ok4 := s.(SizeLister)
+ _, ok5 := s.(PageLister)
+ _, ok6 := s.(ModLister)
+ if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6) {
+ t.Fatal("index-enabled DirStore lost an extension")
+ }
+ var s3 Store = &S3Store{}
+ _, ok1 = s3.(ConditionalDeleter)
+ _, ok2 = s3.(PutStreamer)
+ _, ok3 = s3.(Opener)
+ _, ok4 = s3.(SizeLister)
+ _, ok5 = s3.(PageLister)
+ _, ok6 = s3.(ModLister)
+ if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6) {
+ t.Fatal("S3Store lost an extension")
+ }
+}
+
+// TestPurgeBumpsBounded: a purge deletes many keys under one app. Each delete
+// is its own mutation and costs exactly one index CAS, so the generation lands
+// on the delete count and the write amplification is one put per delete.
+func TestPurgeBumpsBounded(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := indexedDirStore(t)
+ const n = 300
+ for i := 0; i < n; i++ {
+ if err := d.Delete(ctx, fmt.Sprintf("apps/bab/shop/versions/%03d.json", i)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if got := scopeGen(t, d, bumpTestScope); got != n {
+ t.Fatalf("gen = %d after %d deletes, want %d", got, n, n)
+ }
+ // One bump is one CAS: count the index puts the hook issues over the raw backend.
+ var puts atomic.Int64
+ put := func(c context.Context, k string, b []byte, m string) (string, error) {
+ puts.Add(1)
+ return d.putRaw(c, k, b, m)
+ }
+ for i := 0; i < 20; i++ {
+ bumpAfterMutation(ctx, true, bumpTestKey, d.Get, put)
+ }
+ if got := puts.Load(); got != 20 {
+ t.Fatalf("20 bumps issued %d index puts, want exactly one each", got)
+ }
+}
+
+// logTo captures the standard logger for one serial test.
+func logTo(t *testing.T, w *bytes.Buffer) {
+ t.Helper()
+ log.SetOutput(w)
+ t.Cleanup(func() { log.SetOutput(os.Stderr) })
+}
diff --git a/internal/release/conditional_delete.go b/internal/release/conditional_delete.go
index 67917e6a6..290104747 100644
--- a/internal/release/conditional_delete.go
+++ b/internal/release/conditional_delete.go
@@ -60,7 +60,15 @@ func (p prefixStore) DeleteIfMatch(ctx context.Context, key, etag string) error
return DeleteIfMatch(ctx, p.Store, p.full(key), etag)
}
-func (d *DirStore) DeleteIfMatch(_ context.Context, key, etag string) error {
+func (d *DirStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ err := d.deleteIfMatchRaw(ctx, key, etag)
+ if err == nil {
+ d.bump(ctx, key)
+ }
+ return err
+}
+
+func (d *DirStore) deleteIfMatchRaw(_ context.Context, key, etag string) error {
if !exactDeleteETag(etag) {
return ErrInvalidETag
}
@@ -84,6 +92,14 @@ func (d *DirStore) DeleteIfMatch(_ context.Context, key, etag string) error {
}
func (s *S3Store) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ err := s.deleteIfMatchRaw(ctx, key, etag)
+ if err == nil {
+ s.bump(ctx, key)
+ }
+ return err
+}
+
+func (s *S3Store) deleteIfMatchRaw(ctx context.Context, key, etag string) error {
if !exactDeleteETag(etag) {
return ErrInvalidETag
}
diff --git a/internal/release/store.go b/internal/release/store.go
index 3a836d25a..6a77da877 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -340,8 +340,9 @@ var (
// temp-file + rename writes: single-process fidelity, which is what the
// tests and a one-box deployment need.
type DirStore struct {
- root string
- mu sync.Mutex
+ root string
+ mu sync.Mutex
+ changeIndex bool // EnableChangeIndex: bump control/changes.json after each mutation
}
func NewDirStore(root string) *DirStore { return &DirStore{root: root} }
@@ -411,7 +412,24 @@ func (d *DirStore) Open(_ context.Context, key string) (io.ReadCloser, string, e
return f, "", nil
}
-func (d *DirStore) Put(_ context.Context, key string, body []byte, ifMatch string) (string, error) {
+// EnableChangeIndex makes every successful Put, PutFrom, Delete and
+// DeleteIfMatch inside a change scope bump control/changes.json afterwards
+// (design §3.3). Call it before the store is shared; it is not synchronised.
+func (d *DirStore) EnableChangeIndex() { d.changeIndex = true }
+
+func (d *DirStore) bump(ctx context.Context, key string) {
+ bumpAfterMutation(ctx, d.changeIndex, key, d.Get, d.putRaw)
+}
+
+func (d *DirStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ etag, err := d.putRaw(ctx, key, body, ifMatch)
+ if err == nil {
+ d.bump(ctx, key)
+ }
+ return etag, err
+}
+
+func (d *DirStore) putRaw(_ context.Context, key string, body []byte, ifMatch string) (string, error) {
p, err := d.path(key)
if err != nil {
return "", err
@@ -453,7 +471,15 @@ func (d *DirStore) Put(_ context.Context, key string, body []byte, ifMatch strin
// PutFrom is Put streamed: the body goes to a temp file (hashed as it
// goes, so the etag is the same content sha256 Put would give) and is
// renamed into place under d.mu after the same CAS gate.
-func (d *DirStore) PutFrom(_ context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
+func (d *DirStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
+ etag, err := d.putFromRaw(ctx, key, body, size, ifMatch)
+ if err == nil {
+ d.bump(ctx, key)
+ }
+ return etag, err
+}
+
+func (d *DirStore) putFromRaw(_ context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
p, err := d.path(key)
if err != nil {
return "", err
@@ -657,7 +683,15 @@ func (d *DirStore) ListDirs(_ context.Context, prefix string) ([]string, error)
return out, nil
}
-func (d *DirStore) Delete(_ context.Context, key string) error {
+func (d *DirStore) Delete(ctx context.Context, key string) error {
+ err := d.deleteRaw(ctx, key)
+ if err == nil {
+ d.bump(ctx, key)
+ }
+ return err
+}
+
+func (d *DirStore) deleteRaw(_ context.Context, key string) error {
p, err := d.path(key)
if err != nil {
return err
@@ -681,6 +715,8 @@ func (d *DirStore) Delete(_ context.Context, key string) error {
type S3Store struct {
Client *awss3.Client
Bucket string
+
+ changeIndex bool // EnableChangeIndex: bump control/changes.json after each mutation
}
// NewS3Store uses the standard AWS chain for non-root callers. Root never loads
@@ -784,7 +820,46 @@ func (s *S3Store) Open(ctx context.Context, key string) (io.ReadCloser, string,
return out.Body, aws.ToString(out.ETag), nil
}
+// EnableChangeIndex makes every successful Put, PutFrom, Delete and
+// DeleteIfMatch inside a change scope bump control/changes.json afterwards
+// (design §3.3). Call it before the store is shared; it is not synchronised.
+func (s *S3Store) EnableChangeIndex() { s.changeIndex = true }
+
+func (s *S3Store) bump(ctx context.Context, key string) {
+ bumpAfterMutation(ctx, s.changeIndex, key, s.Get, s.putRaw)
+}
+
+// NewDeploymentS3Store is NewS3Store for the deployment bucket: the returned
+// store bumps the change index after each mutation. Stores over any other
+// bucket keep using NewS3Store.
+func NewDeploymentS3Store(ctx context.Context, bucket string) (*S3Store, error) {
+ s, err := NewS3Store(ctx, bucket)
+ if err != nil {
+ return nil, err
+ }
+ s.EnableChangeIndex()
+ return s, nil
+}
+
+// NewDeploymentS3StoreIn is NewS3StoreIn with the change index enabled.
+func NewDeploymentS3StoreIn(ctx context.Context, bucket, region string) (*S3Store, error) {
+ s, err := NewS3StoreIn(ctx, bucket, region)
+ if err != nil {
+ return nil, err
+ }
+ s.EnableChangeIndex()
+ return s, nil
+}
+
func (s *S3Store) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ etag, err := s.putRaw(ctx, key, body, ifMatch)
+ if err == nil {
+ s.bump(ctx, key)
+ }
+ return etag, err
+}
+
+func (s *S3Store) putRaw(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
in := &awss3.PutObjectInput{
Bucket: &s.Bucket,
Key: &key,
@@ -818,6 +893,14 @@ func (s *S3Store) Put(ctx context.Context, key string, body []byte, ifMatch stri
// PutFrom is Put with a streamed, seekable body and an explicit length.
func (s *S3Store) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
+ etag, err := s.putFromRaw(ctx, key, body, size, ifMatch)
+ if err == nil {
+ s.bump(ctx, key)
+ }
+ return etag, err
+}
+
+func (s *S3Store) putFromRaw(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
in := &awss3.PutObjectInput{
Bucket: &s.Bucket,
Key: &key,
@@ -968,6 +1051,14 @@ func (s *S3Store) ListDirs(ctx context.Context, prefix string) ([]string, error)
}
func (s *S3Store) Delete(ctx context.Context, key string) error {
+ err := s.deleteRaw(ctx, key)
+ if err == nil {
+ s.bump(ctx, key)
+ }
+ return err
+}
+
+func (s *S3Store) deleteRaw(ctx context.Context, key string) error {
_, err := s.Client.DeleteObject(ctx, &awss3.DeleteObjectInput{
Bucket: aws.String(s.Bucket),
Key: aws.String(key),