amadan

carlos / platform Public

release: conditional GET and ETag listings, forwarded through prefix stores and the hostops boundary (#336)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude 34cfaf977af36b390d9474fc846a502c992b54c0 parent 559295c
7 files changed, +649 −0
  • .amadan/ci.d/30-minio-contract +30 −0
  • internal/hostops/mutation_boundary.go +15 −0
  • internal/hostops/mutation_boundary_test.go +68 −0
  • internal/hostops/record_writes.go +15 −0
  • internal/release/readext_test.go +178 −0
  • internal/release/s3_minio_test.go +168 −0
  • internal/release/store.go +175 −0
diff --git a/.amadan/ci.d/30-minio-contract b/.amadan/ci.d/30-minio-contract
index a0834406b..80f37806d 100755
--- a/.amadan/ci.d/30-minio-contract
+++ b/.amadan/ci.d/30-minio-contract
@@ -388,4 +388,34 @@ if ! grep -q '^--- PASS: TestT4StorageProbeMinio (' "$out2"; then
exit 1
fi
+# The change index (S3 change-index design §3, §6): release.S3Store against
+# the same server. A 304 on a current ETag, an ETag listing that agrees with
+# Get, and the index bump through every mutation path are the server's
+# behaviour; no fake client shows them. Same rule again: the test skips
+# without CARLOS_MINIO_TEST, so its PASS line has to appear.
+out3=$rundir/test-s3store.log
+set +e
+CARLOS_MINIO_TEST=1 \
+ MINIO_ENDPOINT="http://127.0.0.1:$port" \
+ MINIO_SCOPED_KEY=scopeduser \
+ MINIO_SCOPED_SECRET=scopedsecret1 \
+ MINIO_BUCKET=contract-bucket \
+ go test ./internal/release/ -run TestS3StoreMinioContract -count=1 -v >"$out3" 2>&1
+test_status=$?
+set -e
+
+cat "$out3"
+
+if [ "$test_status" -ne 0 ]; then
+ exit "$test_status"
+fi
+
+if ! grep -q '^--- PASS: TestS3StoreMinioContract (' "$out3"; then
+ echo >&2
+ echo "the S3Store contract test did not PASS: it was skipped, filtered out, or renamed." >&2
+ echo "It skips unless CARLOS_MINIO_TEST=1. Check the env block above and that" >&2
+ echo "TestS3StoreMinioContract still exists in internal/release." >&2
+ exit 1
+fi
+
echo "minio contract: PASS"
diff --git a/internal/hostops/mutation_boundary.go b/internal/hostops/mutation_boundary.go
index 0d4e3661f..b174dd502 100644
--- a/internal/hostops/mutation_boundary.go
+++ b/internal/hostops/mutation_boundary.go
@@ -50,6 +50,8 @@ var (
_ release.ModLister = (*mutationBoundary)(nil)
_ release.SizeLister = (*mutationBoundary)(nil)
_ release.PageLister = (*mutationBoundary)(nil)
+ _ release.ConditionalGetter = (*mutationBoundary)(nil)
+ _ release.ETagLister = (*mutationBoundary)(nil)
_ release.BucketListDirs = (*mutationBoundary)(nil)
)
@@ -137,6 +139,19 @@ func (s *mutationBoundary) ListPage(ctx context.Context, prefix, startAfter stri
return nil, ErrContract
}
+func (s *mutationBoundary) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ if g, ok := s.raw.(release.ConditionalGetter); ok {
+ return g.GetIfChanged(ctx, key, etag)
+ }
+ return nil, "", false, ErrContract
+}
+func (s *mutationBoundary) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
+ if l, ok := s.raw.(release.ETagLister); ok {
+ return l.ListETags(ctx, prefix)
+ }
+ return nil, ErrContract
+}
+
// ---- mutations: classified, then exact ----
func (s *mutationBoundary) Put(ctx context.Context, key string, body []byte, tag string) (string, error) {
diff --git a/internal/hostops/mutation_boundary_test.go b/internal/hostops/mutation_boundary_test.go
index a6d87e848..0776255c6 100644
--- a/internal/hostops/mutation_boundary_test.go
+++ b/internal/hostops/mutation_boundary_test.go
@@ -77,3 +77,71 @@ func TestMutationBoundaryDisabledIsOriginalHandle(t *testing.T) {
t.Fatal("enabled without production authority")
}
}
+
+func TestBoundaryForwardsReadExtensions(t *testing.T) {
+ ctx := t.Context()
+ raw := release.NewDirStore(t.TempDir())
+ const key = "apps/bab/demo/a.json"
+ if _, err := raw.Put(ctx, key, []byte("a"), ""); err != nil {
+ t.Fatal(err)
+ }
+ // The production chain: boundary over the record store over the bucket.
+ records, err := NewConditionalRecordStore(raw, make([]byte, 32))
+ if err != nil {
+ t.Fatal(err)
+ }
+ guarded, err := NewMutationBoundary(true, records, func(context.Context, MutationTarget) error { return nil })
+ if err != nil {
+ t.Fatal(err)
+ }
+ cg, ok := guarded.(release.ConditionalGetter)
+ if !ok {
+ t.Fatal("boundary hides ConditionalGetter")
+ }
+ el, ok := guarded.(release.ETagLister)
+ if !ok {
+ t.Fatal("boundary hides ETagLister")
+ }
+ body, e, changed, err := cg.GetIfChanged(ctx, key, "")
+ if err != nil || !changed || string(body) != "a" {
+ t.Fatalf("GetIfChanged: %q %v %v", body, changed, err)
+ }
+ if _, _, changed, err = cg.GetIfChanged(ctx, key, e); err != nil || changed {
+ t.Fatalf("unchanged: %v %v", changed, err)
+ }
+ got, err := el.ListETags(ctx, "apps/")
+ if err != nil || len(got) != 1 || got[0].Key != key || got[0].ETag != e {
+ t.Fatalf("ListETags: %v %v", got, err)
+ }
+}
+
+// bareStore has none of the optional extensions.
+type bareStore struct{ release.Store }
+
+func TestBoundaryWithoutReadExtensionsIsContractError(t *testing.T) {
+ ctx := t.Context()
+ guarded, err := NewMutationBoundary(true, bareStore{release.NewDirStore(t.TempDir())}, func(context.Context, MutationTarget) error { return nil })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, _, _, err = guarded.(release.ConditionalGetter).GetIfChanged(ctx, "apps/bab/demo/a.json", ""); !errors.Is(err, ErrContract) {
+ t.Fatalf("GetIfChanged: %v", err)
+ }
+ if _, err = guarded.(release.ETagLister).ListETags(ctx, "apps/"); !errors.Is(err, ErrContract) {
+ t.Fatalf("ListETags: %v", err)
+ }
+}
+
+func TestRecordStoreWithoutReadExtensionsIsContractError(t *testing.T) {
+ ctx := t.Context()
+ records, err := NewConditionalRecordStore(bareStore{release.NewDirStore(t.TempDir())}, make([]byte, 32))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, _, _, err = records.(release.ConditionalGetter).GetIfChanged(ctx, "apps/bab/demo/a.json", ""); !errors.Is(err, ErrContract) {
+ t.Fatalf("GetIfChanged: %v", err)
+ }
+ if _, err = records.(release.ETagLister).ListETags(ctx, "apps/"); !errors.Is(err, ErrContract) {
+ t.Fatalf("ListETags: %v", err)
+ }
+}
diff --git a/internal/hostops/record_writes.go b/internal/hostops/record_writes.go
index 129f48d85..5197f09dd 100644
--- a/internal/hostops/record_writes.go
+++ b/internal/hostops/record_writes.go
@@ -33,6 +33,8 @@ var (
_ release.ModLister = conditionalRecordStore{}
_ release.SizeLister = conditionalRecordStore{}
_ release.PageLister = conditionalRecordStore{}
+ _ release.ConditionalGetter = conditionalRecordStore{}
+ _ release.ETagLister = conditionalRecordStore{}
_ release.BucketListDirs = conditionalRecordStore{}
)
@@ -83,6 +85,19 @@ func (s conditionalRecordStore) ListPage(ctx context.Context, prefix, startAfter
return nil, ErrContract
}
+func (s conditionalRecordStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ if g, ok := s.raw.(release.ConditionalGetter); ok {
+ return g.GetIfChanged(ctx, key, etag)
+ }
+ return nil, "", false, ErrContract
+}
+func (s conditionalRecordStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
+ if l, ok := s.raw.(release.ETagLister); ok {
+ return l.ListETags(ctx, prefix)
+ }
+ return nil, ErrContract
+}
+
func (s conditionalRecordStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
if !provisioningKey(key) {
return s.raw.Put(ctx, key, body, etag)
diff --git a/internal/release/readext_test.go b/internal/release/readext_test.go
new file mode 100644
index 000000000..d0905ec26
--- /dev/null
+++ b/internal/release/readext_test.go
@@ -0,0 +1,178 @@
+package release
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "testing"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+func TestGetIfChangedDirStore(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := NewDirStore(t.TempDir())
+ const key = "apps/bab/shop/instances/h.json"
+ if _, err := d.Put(ctx, key, []byte("one"), ""); err != nil {
+ t.Fatal(err)
+ }
+ var cg ConditionalGetter = d
+ body, e, changed, err := cg.GetIfChanged(ctx, key, "")
+ if err != nil || !changed || string(body) != "one" || e == "" {
+ t.Fatalf("empty etag: %q %q %v %v", body, e, changed, err)
+ }
+ _, ge, err := d.Get(ctx, key)
+ if err != nil || ge != e {
+ t.Fatalf("GetIfChanged etag %q, Get etag %q (%v)", e, ge, err)
+ }
+ body, e2, changed, err := cg.GetIfChanged(ctx, key, e)
+ if err != nil || changed || body != nil || e2 != e {
+ t.Fatalf("unchanged: %q %q %v %v", body, e2, changed, err)
+ }
+ if _, err := d.Put(ctx, key, []byte("two"), e); err != nil {
+ t.Fatal(err)
+ }
+ body, e3, changed, err := cg.GetIfChanged(ctx, key, e)
+ if err != nil || !changed || string(body) != "two" || e3 == e {
+ t.Fatalf("rewritten: %q %q %v %v", body, e3, changed, err)
+ }
+}
+
+func TestGetIfChangedAbsent(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := NewDirStore(t.TempDir())
+ for _, etag := range []string{"", "some-etag"} {
+ _, _, changed, err := d.GetIfChanged(ctx, "apps/bab/shop/absent.json", etag)
+ if !errors.Is(err, os.ErrNotExist) || changed {
+ t.Fatalf("etag %q: changed=%v err=%v", etag, changed, err)
+ }
+ }
+}
+
+func TestListETagsMatchesGet(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := NewDirStore(t.TempDir())
+ keys := []string{"apps/bab/shop/b.json", "apps/bab/shop/a.json", "apps/bab/other/c.json", "accounts/bab/x.json"}
+ for i, k := range keys {
+ if _, err := d.Put(ctx, k, []byte{byte('a' + i)}, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ var el ETagLister = d
+ got, err := el.ListETags(ctx, "apps/bab/")
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := []string{"apps/bab/other/c.json", "apps/bab/shop/a.json", "apps/bab/shop/b.json"}
+ if len(got) != len(want) {
+ t.Fatalf("listed %v, want keys %v", got, want)
+ }
+ for i, ke := range got {
+ if ke.Key != want[i] {
+ t.Fatalf("entry %d: %q, want %q (sorted by key)", i, ke.Key, want[i])
+ }
+ _, e, err := d.Get(ctx, ke.Key)
+ if err != nil || e != ke.ETag {
+ t.Fatalf("%s: listed etag %q, Get etag %q (%v)", ke.Key, ke.ETag, e, err)
+ }
+ }
+ none, err := d.ListETags(ctx, "nothing/")
+ if err != nil || len(none) != 0 {
+ t.Fatalf("empty prefix: %v %v", none, err)
+ }
+}
+
+func TestPrefixStoreForwardsReadExtensions(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ inner := NewDirStore(t.TempDir())
+ if _, err := inner.Put(ctx, "fleets/f/apps/bab/shop/x.json", []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := inner.Put(ctx, "fleets/g/apps/bab/shop/y.json", []byte("y"), ""); err != nil {
+ t.Fatal(err)
+ }
+ s := NewPrefixStore(inner, "fleets/f/")
+ cg, ok := s.(ConditionalGetter)
+ if !ok {
+ t.Fatal("prefix store does not forward ConditionalGetter")
+ }
+ el, ok := s.(ETagLister)
+ if !ok {
+ t.Fatal("prefix store does not forward ETagLister")
+ }
+ body, e, changed, err := cg.GetIfChanged(ctx, "apps/bab/shop/x.json", "")
+ if err != nil || !changed || string(body) != "x" {
+ t.Fatalf("GetIfChanged: %q %v %v", body, changed, err)
+ }
+ if _, _, changed, err = cg.GetIfChanged(ctx, "apps/bab/shop/x.json", e); err != nil || changed {
+ t.Fatalf("unchanged through prefix: %v %v", changed, err)
+ }
+ if _, _, _, err = cg.GetIfChanged(ctx, "apps/bab/shop/y.json", ""); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("other fleet's key must be absent: %v", err)
+ }
+ got, err := el.ListETags(ctx, "apps/")
+ if err != nil || len(got) != 1 || got[0].Key != "apps/bab/shop/x.json" || got[0].ETag != e {
+ t.Fatalf("ListETags: %v %v", got, err)
+ }
+}
+
+// bareStore has none of the optional extensions.
+type bareStore struct{ Store }
+
+func TestPrefixStoreWithoutExtensionsRefuses(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ s := NewPrefixStore(bareStore{NewDirStore(t.TempDir())}, "fleets/f/")
+ if _, _, _, err := s.(ConditionalGetter).GetIfChanged(ctx, "k", ""); !errors.Is(err, errors.ErrUnsupported) {
+ t.Fatalf("GetIfChanged: %v", err)
+ }
+ if _, err := s.(ETagLister).ListETags(ctx, "k"); !errors.Is(err, errors.ErrUnsupported) {
+ t.Fatalf("ListETags: %v", err)
+ }
+}
+
+// The SDK reports a 304 from GetObject as an error; GetIfChanged must read it
+// as "unchanged". A loopback server stands in for the 304 so the mapping runs
+// here, not only against MinIO in CI.
+func TestS3GetIfChangedMaps304(t *testing.T) {
+ t.Parallel()
+ const etag = `"abc123"`
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Header.Get("If-None-Match") == etag {
+ w.WriteHeader(http.StatusNotModified)
+ return
+ }
+ w.Header().Set("ETag", etag)
+ _, _ = w.Write([]byte("body"))
+ }))
+ defer srv.Close()
+ client := awss3.New(awss3.Options{
+ Region: "us-east-1",
+ Credentials: credentials.NewStaticCredentialsProvider("k", "s", ""),
+ BaseEndpoint: aws.String(srv.URL),
+ UsePathStyle: true,
+ Retryer: aws.NopRetryer{},
+ })
+ s := &S3Store{Client: client, Bucket: "b"}
+ ctx := context.Background()
+ body, e, changed, err := s.GetIfChanged(ctx, "k", "")
+ if err != nil || !changed || string(body) != "body" || e != etag {
+ t.Fatalf("empty etag: %q %q %v %v", body, e, changed, err)
+ }
+ body, e, changed, err = s.GetIfChanged(ctx, "k", etag)
+ if err != nil || changed || body != nil || e != etag {
+ t.Fatalf("304: %q %q %v %v", body, e, changed, err)
+ }
+ body, e, changed, err = s.GetIfChanged(ctx, "k", `"other"`)
+ if err != nil || !changed || string(body) != "body" {
+ t.Fatalf("stale etag: %q %q %v %v", body, e, changed, err)
+ }
+}
diff --git a/internal/release/s3_minio_test.go b/internal/release/s3_minio_test.go
new file mode 100644
index 000000000..23d34442f
--- /dev/null
+++ b/internal/release/s3_minio_test.go
@@ -0,0 +1,168 @@
+package release
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// TestS3StoreMinioContract runs release.S3Store against a real S3-compatible
+// server: the conditional GET (a 304 for the current ETag), the ETag listing
+// agreeing with Get, and the change index bump through every S3Store
+// mutation path. A fake client cannot show any of these: the 304 and the
+// ETag forms are the server's. Skipped unless CARLOS_MINIO_TEST=1;
+// .amadan/ci.d/30-minio-contract stands the server up and requires this
+// test's PASS line, so a skip cannot pass for coverage there.
+func TestS3StoreMinioContract(t *testing.T) {
+ if os.Getenv("CARLOS_MINIO_TEST") != "1" {
+ t.Skip("set CARLOS_MINIO_TEST=1 (with a running Minio and the MINIO_* env vars) to run the S3Store contract test; see .amadan/ci.d/30-minio-contract, which stands one up")
+ }
+ endpoint := minioEnv(t, "MINIO_ENDPOINT")
+ key := minioEnv(t, "MINIO_SCOPED_KEY")
+ secret := minioEnv(t, "MINIO_SCOPED_SECRET")
+ bucket := minioEnv(t, "MINIO_BUCKET")
+
+ client := awss3.New(awss3.Options{
+ Region: "us-east-1",
+ Credentials: credentials.NewStaticCredentialsProvider(key, secret, ""),
+ BaseEndpoint: aws.String(endpoint),
+ UsePathStyle: true,
+ RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+ ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
+ })
+ s := &S3Store{Client: client, Bucket: bucket}
+ s.EnableChangeIndex()
+ ctx := context.Background()
+
+ // One scope per run, so nothing else in the shared bucket moves its
+ // generation. control/changes.json is the one shared key: the test
+ // compares generations before and after and leaves it in place.
+ app := fmt.Sprintf("mt%d", time.Now().UnixNano())
+ scope := "apps/" + app + "/shop"
+ base := scope + "/"
+ var created []string
+ t.Cleanup(func() {
+ for _, k := range created {
+ _ = s.deleteRaw(context.Background(), k) // not Delete: no bump on the way out
+ }
+ })
+ keyA, keyB := base+"a.json", base+"b.json"
+ created = append(created, keyA, keyB)
+
+ gen := func() uint64 {
+ t.Helper()
+ idx, _, err := ReadChangeIndex(ctx, s) // an absent index reads as empty
+ if err != nil {
+ t.Fatalf("read change index: %v", err)
+ }
+ return idx.Entries[scope]
+ }
+ bumped := func(name string, before uint64) uint64 {
+ t.Helper()
+ after := gen()
+ if after != before+1 {
+ t.Fatalf("%s: generation %d -> %d, want exactly +1", name, before, after)
+ }
+ return after
+ }
+
+ t.Run("conditional get and etag listing", func(t *testing.T) {
+ etag, err := s.putRaw(ctx, keyA, []byte("one"), "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, e, changed, err := s.GetIfChanged(ctx, keyA, "")
+ if err != nil || !changed || string(body) != "one" || e != etag {
+ t.Fatalf("empty etag: %q %q (put %q) %v %v", body, e, etag, changed, err)
+ }
+ body, e2, changed, err := s.GetIfChanged(ctx, keyA, e)
+ if err != nil || changed || body != nil || e2 != e {
+ t.Fatalf("current etag must be a 304: %q %q %v %v", body, e2, changed, err)
+ }
+ if _, err := s.putRaw(ctx, keyA, []byte("two"), e); err != nil {
+ t.Fatal(err)
+ }
+ body, e3, changed, err := s.GetIfChanged(ctx, keyA, e)
+ if err != nil || !changed || string(body) != "two" || e3 == e {
+ t.Fatalf("after rewrite: %q %q %v %v", body, e3, changed, err)
+ }
+ if _, _, _, err := s.GetIfChanged(ctx, base+"absent.json", e); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("absent: %v", err)
+ }
+ if _, err := s.putRaw(ctx, keyB, []byte("b"), ""); err != nil {
+ t.Fatal(err)
+ }
+ listed, err := s.ListETags(ctx, base)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(listed) != 2 || listed[0].Key != keyA || listed[1].Key != keyB {
+ t.Fatalf("listed %v, want %s and %s sorted", listed, keyA, keyB)
+ }
+ for _, ke := range listed {
+ _, ge, err := s.Get(ctx, ke.Key)
+ if err != nil || ge != ke.ETag {
+ t.Fatalf("%s: listed etag %q, Get etag %q (%v)", ke.Key, ke.ETag, ge, err)
+ }
+ }
+ for _, k := range []string{keyA, keyB} {
+ if err := s.deleteRaw(ctx, k); err != nil {
+ t.Fatal(err)
+ }
+ }
+ })
+
+ t.Run("bump through every mutation path", func(t *testing.T) {
+ g := gen()
+ etag, err := s.Put(ctx, keyA, []byte("one"), "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ g = bumped("Put", g)
+
+ // A Put whose ifMatch is stale loses and must not bump.
+ if _, err := s.Put(ctx, keyA, []byte("stale"), `"00000000000000000000000000000000"`); !errors.Is(err, ErrCASConflict) {
+ t.Fatalf("stale Put: %v", err)
+ }
+ if after := gen(); after != g {
+ t.Fatalf("a refused Put moved the generation %d -> %d", g, after)
+ }
+
+ etag, err = s.PutFrom(ctx, keyA, bytes.NewReader([]byte("two")), 3, etag)
+ if err != nil {
+ t.Fatal(err)
+ }
+ g = bumped("PutFrom", g)
+
+ if err := DeleteIfMatch(ctx, s, keyA, etag); err != nil {
+ t.Fatal(err)
+ }
+ g = bumped("DeleteIfMatch", g)
+
+ if _, err := s.Put(ctx, keyA, []byte("three"), ""); err != nil {
+ t.Fatal(err)
+ }
+ g = bumped("Put again", g)
+ if err := s.Delete(ctx, keyA); err != nil {
+ t.Fatal(err)
+ }
+ bumped("Delete", g)
+ })
+}
+
+func minioEnv(t *testing.T, name string) string {
+ t.Helper()
+ v := os.Getenv(name)
+ if v == "" {
+ t.Fatalf("%s is not set (required when CARLOS_MINIO_TEST=1)", name)
+ }
+ return v
+}
diff --git a/internal/release/store.go b/internal/release/store.go
index 6a77da877..5f7d76e09 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -206,6 +206,64 @@ var (
_ ModLister = (*S3Store)(nil)
)
+// ConditionalGetter is the optional conditional read: GetIfChanged is Get
+// unless the object still carries etag, in which case it returns
+// nil, etag, false, nil without the body (S3 answers 304). An empty etag
+// behaves as Get. An absent object returns an os.ErrNotExist-wrapped error.
+// Optional and type-asserted, like ModLister: a store without it is read the
+// old way.
+type ConditionalGetter interface {
+ GetIfChanged(ctx context.Context, key, etag string) (body []byte, newETag string, changed bool, err error)
+}
+
+// KeyETag pairs a listed key with its ETag.
+type KeyETag struct {
+ Key string
+ ETag string
+}
+
+// ETagLister is the optional listing that carries each object's ETag, sorted
+// by key, in the same form Get returns it (S3 keeps its quoting, DirStore its
+// content hash), so a listed ETag can be compared with one read earlier.
+type ETagLister interface {
+ ListETags(ctx context.Context, prefix string) ([]KeyETag, error)
+}
+
+// Both real stores and the prefix wrapper answer both; only fakes fall back.
+var (
+ _ ConditionalGetter = (*DirStore)(nil)
+ _ ConditionalGetter = (*S3Store)(nil)
+ _ ConditionalGetter = prefixStore{}
+ _ ETagLister = (*DirStore)(nil)
+ _ ETagLister = (*S3Store)(nil)
+ _ ETagLister = prefixStore{}
+)
+
+// GetIfChanged rebases the key, like Get. A raw store without the extension
+// answers errors.ErrUnsupported, which callers read as "unknown".
+func (p prefixStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ if g, ok := p.Store.(ConditionalGetter); ok {
+ return g.GetIfChanged(ctx, p.full(key), etag)
+ }
+ return nil, "", false, errors.ErrUnsupported
+}
+
+// ListETags rebases the query prefix and strips the base back off, like List.
+func (p prefixStore) ListETags(ctx context.Context, prefix string) ([]KeyETag, error) {
+ l, ok := p.Store.(ETagLister)
+ if !ok {
+ return nil, errors.ErrUnsupported
+ }
+ got, err := l.ListETags(ctx, p.full(prefix))
+ if err != nil {
+ return nil, err
+ }
+ for i := range got {
+ got[i].Key = strings.TrimPrefix(got[i].Key, p.base)
+ }
+ return got, nil
+}
+
// KeySize is one object and its byte size.
type KeySize struct {
Key string
@@ -602,6 +660,60 @@ func (d *DirStore) ListMod(_ context.Context, prefix string) ([]KeyMod, error) {
return out, err
}
+// GetIfChanged is Get, except an object whose content hash is etag is not
+// returned.
+func (d *DirStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ b, e, err := d.Get(ctx, key)
+ if err != nil {
+ return nil, "", false, err
+ }
+ if etag != "" && e == etag {
+ return nil, etag, false, nil
+ }
+ return b, e, true, nil
+}
+
+// ListETags is ListMod's walk with each file's content hash alongside,
+// key-sorted. The hash is the one Get returns.
+func (d *DirStore) ListETags(_ context.Context, prefix string) ([]KeyETag, error) {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ var out []KeyETag
+ err := filepath.WalkDir(d.root, func(p string, de fs.DirEntry, err error) error {
+ if err != nil {
+ if errors.Is(err, fs.ErrNotExist) {
+ return nil // an empty store lists empty, like an empty bucket
+ }
+ return err
+ }
+ if de.IsDir() && strings.HasPrefix(de.Name(), ".carlos-write-") {
+ return filepath.SkipDir
+ }
+ if de.IsDir() || strings.HasSuffix(p, ".tmp") {
+ return nil
+ }
+ rel, err := filepath.Rel(d.root, p)
+ if err != nil {
+ return err
+ }
+ key := filepath.ToSlash(rel)
+ if !strings.HasPrefix(key, prefix) {
+ return nil
+ }
+ etag, err := hashFile(p)
+ if err != nil {
+ if errors.Is(err, fs.ErrNotExist) {
+ return nil // deleted by another process mid-walk: not listed
+ }
+ return err
+ }
+ out = append(out, KeyETag{Key: key, ETag: etag})
+ return nil
+ })
+ sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
+ return out, err
+}
+
// ListSizes mirrors ListMod's walk, collecting each file's byte size instead
// of its mtime.
func (d *DirStore) ListSizes(_ context.Context, prefix string) ([]KeySize, error) {
@@ -999,6 +1111,69 @@ func (s *S3Store) ListMod(ctx context.Context, prefix string) ([]KeyMod, error)
return out, nil
}
+// s3NotModified recognises the 304 a conditional GET answers with. The SDK
+// surfaces it as an error: by HTTP status, or by the NotModified code an
+// S3-compatible may set instead.
+func s3NotModified(err error) bool {
+ var httpErr interface{ HTTPStatusCode() int }
+ if errors.As(err, &httpErr) && httpErr.HTTPStatusCode() == http.StatusNotModified {
+ return true
+ }
+ var apiErr smithy.APIError
+ if errors.As(err, &apiErr) {
+ switch apiErr.ErrorCode() {
+ case "NotModified", "304":
+ return true
+ }
+ }
+ return false
+}
+
+// GetIfChanged sends If-None-Match, so an unchanged object costs a 304 and no
+// body.
+func (s *S3Store) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ in := &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key}
+ if etag != "" {
+ in.IfNoneMatch = &etag
+ }
+ out, err := s.Client.GetObject(ctx, in)
+ if err != nil {
+ if etag != "" && s3NotModified(err) {
+ return nil, etag, false, nil
+ }
+ if s3NotFound(err) {
+ return nil, "", false, fmt.Errorf("s3store: %s: %w", key, os.ErrNotExist)
+ }
+ return nil, "", false, err
+ }
+ defer out.Body.Close()
+ b, err := io.ReadAll(out.Body)
+ if err != nil {
+ return nil, "", false, err
+ }
+ return b, aws.ToString(out.ETag), true, nil
+}
+
+// ListETags pages ListObjectsV2 and keeps each ETag exactly as the listing
+// sends it, which is the form GetObject reports.
+func (s *S3Store) ListETags(ctx context.Context, prefix string) ([]KeyETag, error) {
+ var out []KeyETag
+ p := awss3.NewListObjectsV2Paginator(s.Client, &awss3.ListObjectsV2Input{
+ Bucket: &s.Bucket,
+ Prefix: &prefix,
+ })
+ for p.HasMorePages() {
+ page, err := p.NextPage(ctx)
+ if err != nil {
+ return nil, err
+ }
+ for _, o := range page.Contents {
+ out = append(out, KeyETag{Key: aws.ToString(o.Key), ETag: aws.ToString(o.ETag)})
+ }
+ }
+ return out, nil
+}
+
// ListSizes mirrors ListMod's paginator loop, collecting each object's byte
// size instead of its LastModified stamp.
func (s *S3Store) ListSizes(ctx context.Context, prefix string) ([]KeySize, error) {