amadan

carlos / platform Public

lease.HasReplica and release.NewS3StoreIn for the seed fallback (Plan 4 Task 2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c555b2ea96c172428773b7e0087d8b390dad6d8b parent 0c9b670
4 files changed, +112 −1
  • internal/lease/check.go +41 −0
  • internal/lease/hasreplica_test.go +42 −0
  • internal/release/store.go +16 −1
  • internal/release/store_test.go +13 −0

CI passed — run details

diff --git a/internal/lease/check.go b/internal/lease/check.go
index 28a516964..e118d082a 100644
--- a/internal/lease/check.go
+++ b/internal/lease/check.go
@@ -10,6 +10,8 @@ import (
"os"
"path/filepath"
+ "github.com/benbjohnson/litestream"
+ "github.com/superfly/ltx"
_ "modernc.org/sqlite"
)
@@ -144,3 +146,42 @@ func ChangeCounter(path string) (uint32, error) {
}
return binary.BigEndian.Uint32(header[24:28]), nil
}
+
+// HasReplica reports whether s3://bucket/prefix holds any LTX file at any
+// level: the seed fallback's "has this host a replica of its own" (seeded
+// instances spec §3.4 step 1). A listing error is returned, never read as
+// "no replica".
+func HasReplica(ctx context.Context, bucket, prefix, region string) (bool, error) {
+ region, err := resolveRegion(ctx, region)
+ if err != nil {
+ return false, err
+ }
+ return hasReplicaIn(ctx, newReplicaClient(bucket, prefix, region))
+}
+
+// ltxLister is the one replica-client method HasReplica needs.
+type ltxLister interface {
+ LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)
+}
+
+func hasReplicaIn(ctx context.Context, client ltxLister) (bool, error) {
+ for level := 0; level <= litestream.SnapshotLevel; level++ {
+ itr, err := client.LTXFiles(ctx, level, 0, false)
+ if err != nil {
+ return false, err
+ }
+ found := itr.Next()
+ ierr := itr.Err()
+ cerr := itr.Close()
+ if ierr != nil {
+ return false, ierr
+ }
+ if cerr != nil {
+ return false, cerr
+ }
+ if found {
+ return true, nil
+ }
+ }
+ return false, nil
+}
diff --git a/internal/lease/hasreplica_test.go b/internal/lease/hasreplica_test.go
new file mode 100644
index 000000000..2dff64b41
--- /dev/null
+++ b/internal/lease/hasreplica_test.go
@@ -0,0 +1,42 @@
+package lease
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/superfly/ltx"
+)
+
+type fakeLister struct {
+ levels map[int][]*ltx.FileInfo
+ err error
+ asked []int
+}
+
+func (f *fakeLister) LTXFiles(_ context.Context, level int, _ ltx.TXID, _ bool) (ltx.FileIterator, error) {
+ f.asked = append(f.asked, level)
+ if f.err != nil {
+ return nil, f.err
+ }
+ return ltx.NewFileInfoSliceIterator(f.levels[level]), nil
+}
+
+func TestHasReplicaIn(t *testing.T) {
+ ctx := context.Background()
+ empty := &fakeLister{}
+ if has, err := hasReplicaIn(ctx, empty); has || err != nil {
+ t.Fatalf("empty = %v, %v", has, err)
+ }
+ if len(empty.asked) != 10 {
+ t.Fatalf("levels asked = %v, want 0 to 9", empty.asked)
+ }
+ snap := &fakeLister{levels: map[int][]*ltx.FileInfo{9: {{Level: 9, MinTXID: 1, MaxTXID: 7, Size: 4096}}}}
+ if has, err := hasReplicaIn(ctx, snap); !has || err != nil {
+ t.Fatalf("a snapshot at level 9 = %v, %v", has, err)
+ }
+ broken := &fakeLister{err: errors.New("listing failed")}
+ if has, err := hasReplicaIn(ctx, broken); has || err == nil {
+ t.Fatalf("a failed listing = %v, %v; want an error, never false", has, err)
+ }
+}
diff --git a/internal/release/store.go b/internal/release/store.go
index 57449cf47..8b82407cb 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -685,7 +685,22 @@ type S3Store struct {
// in internal/s3compat, so this store and the hibernation lease cannot drift
// apart against the same bucket. See that package for the quirk table.
func NewS3Store(ctx context.Context, bucket string) (*S3Store, error) {
- cfg, err := awsconfig.LoadDefaultConfig(ctx)
+ return newS3Store(ctx, bucket)
+}
+
+// NewS3StoreIn is NewS3Store with the bucket's region pinned: a box's or a
+// console's data bucket is named with its region beside it
+// (CARLOS_HIBERNATE_REGION), and the default chain's region may be
+// another. An empty region is NewS3Store.
+func NewS3StoreIn(ctx context.Context, bucket, region string) (*S3Store, error) {
+ if region == "" {
+ return newS3Store(ctx, bucket)
+ }
+ return newS3Store(ctx, bucket, awsconfig.WithRegion(region))
+}
+
+func newS3Store(ctx context.Context, bucket string, opts ...func(*awsconfig.LoadOptions) error) (*S3Store, error) {
+ cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...)
if err != nil {
return nil, fmt.Errorf("s3store: load aws config: %w", err)
}
diff --git a/internal/release/store_test.go b/internal/release/store_test.go
index c0bbe99a5..22588c7ea 100644
--- a/internal/release/store_test.go
+++ b/internal/release/store_test.go
@@ -381,3 +381,16 @@ func TestPrefixStoreListDirsRebasesAndStrips(t *testing.T) {
t.Fatalf("ListDirs = %v, want exactly [status/boxes/]", dirs)
}
}
+
+func TestNewS3StoreInPinsTheRegion(t *testing.T) {
+ t.Setenv("AWS_REGION", "us-east-1")
+ t.Setenv("AWS_ACCESS_KEY_ID", "k")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "s")
+ s, err := NewS3StoreIn(context.Background(), "b", "eu-west-1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := s.Client.Options().Region; got != "eu-west-1" {
+ t.Fatalf("region = %q, want eu-west-1", got)
+ }
+}