amadan

carlos / platform Public

lease, console: ranged exact-version reads for resumed door downloads (Plan 7 review round 1)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5d15546b8608b0675af15753156d1b840575e50b parent 87c8ef9
7 files changed, +423 −5
  • internal/console/apidbclaim.go +12 −4
  • internal/console/apidbclaim_range_test.go +195 −0
  • internal/lease/leasetest/bucket.go +27 −0
  • internal/lease/leasetest/contract.go +85 −0
  • internal/lease/leasetest/s3server.go +25 −0
  • internal/lease/versioned.go +28 −1
  • internal/lease/versioned_test.go +51 −0
diff --git a/internal/console/apidbclaim.go b/internal/console/apidbclaim.go
index 9283268a2..964d66466 100644
--- a/internal/console/apidbclaim.go
+++ b/internal/console/apidbclaim.go
@@ -254,16 +254,24 @@ func (v *dbVersionReader) OpenLTXFile(ctx context.Context, level int, lo, hi ltx
if err != nil {
return nil, err
}
- rc, err := v.store.OpenVersion(ctx, v.bucket, v.handle.Key, v.handle.VersionID)
+ var rc io.ReadCloser
+ ranged, ok := v.store.(lease.VersionRangeOpener)
+ if ok && n > 0 {
+ rc, err = ranged.OpenVersionRange(ctx, v.bucket, v.handle.Key, v.handle.VersionID, offset, n)
+ } else {
+ rc, err = v.store.OpenVersion(ctx, v.bucket, v.handle.Key, v.handle.VersionID)
+ }
if errors.Is(err, lease.ErrMirrorNotFound) {
return nil, os.ErrNotExist
}
if err != nil {
return nil, err
}
- if _, err = io.CopyN(io.Discard, rc, offset); err != nil {
- rc.Close()
- return nil, err
+ if !ok {
+ if _, err = io.CopyN(io.Discard, rc, offset); err != nil {
+ rc.Close()
+ return nil, err
+ }
}
return &dbRangeReader{Reader: io.LimitReader(rc, n), Closer: rc}, nil
}
diff --git a/internal/console/apidbclaim_range_test.go b/internal/console/apidbclaim_range_test.go
new file mode 100644
index 000000000..937c44ab6
--- /dev/null
+++ b/internal/console/apidbclaim_range_test.go
@@ -0,0 +1,195 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "os"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+type doorRangeHook struct {
+ doorOpenHook
+ openRange func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error)
+}
+
+func (v doorRangeHook) OpenVersionRange(ctx context.Context, b, k, id string, offset, length int64) (io.ReadCloser, error) {
+ return v.openRange(ctx, b, k, id, offset, length)
+}
+
+type doorCountedReader struct {
+ io.ReadCloser
+ ctx context.Context
+ read *atomic.Int64
+ delay time.Duration
+}
+
+func (r *doorCountedReader) Read(p []byte) (int, error) {
+ if r.delay > 0 {
+ timer := time.NewTimer(r.delay)
+ defer timer.Stop()
+ select {
+ case <-r.ctx.Done():
+ return 0, r.ctx.Err()
+ case <-timer.C:
+ }
+ if len(p) > 4096 {
+ p = p[:4096]
+ }
+ }
+ n, err := r.ReadCloser.Read(p)
+ r.read.Add(int64(n))
+ return n, err
+}
+
+func TestDoorPlanDeepRangeReadsOnlyRequestedBytes(t *testing.T) {
+ for _, ranged := range []bool{false, true} {
+ t.Run(fmt.Sprint(ranged), func(t *testing.T) {
+ s, g, _, model := doorPlanFixture(t, true)
+ const offset = 4 << 20
+ payload := append(bytes.Repeat([]byte("p"), offset), []byte("original")...)
+ key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
+ id := model.Put("data", key, payload)
+ model.Put("data", key, []byte("replacement"))
+ var read atomic.Int64
+ base := doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
+ rc, err := model.OpenVersion(ctx, b, k, v)
+ return &doorCountedReader{ReadCloser: rc, read: &read}, err
+ }}
+ var store lease.VersionedStore = base
+ if ranged {
+ store = doorRangeHook{doorOpenHook: base, openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
+ if b != "data" || k != key || v != id || off != offset || length != 3 {
+ t.Fatalf("range arguments = %s %s %s %d %d", b, k, v, off, length)
+ }
+ rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
+ return &doorCountedReader{ReadCloser: rc, read: &read}, err
+ }}
+ }
+ rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
+ Location: g.claim.Claim.Holder.Location, Key: key, VersionID: id, Size: int64(len(payload)),
+ }}
+ rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, offset, 3)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil || string(body) != "ori" {
+ t.Fatalf("body = %q, %v", body, err)
+ }
+ want := int64(offset + 3)
+ if ranged {
+ want = 3
+ }
+ if got := read.Load(); got != want {
+ t.Fatalf("storage read %d bytes; want %d", got, want)
+ }
+ })
+ }
+}
+
+func TestDoorPlanRangeErrorsNeverFallBack(t *testing.T) {
+ for _, failure := range []error{lease.ErrMirrorNotFound, lease.ErrIsDeleteMarker, context.Canceled, errors.New("offline")} {
+ t.Run(failure.Error(), func(t *testing.T) {
+ s, g, _, model := doorPlanFixture(t, true)
+ key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
+ model.Put("data", key, []byte("replacement"))
+ var fullOpens int
+ store := doorRangeHook{
+ doorOpenHook: doorOpenHook{VersionedStore: model, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
+ fullOpens++
+ return io.NopCloser(bytes.NewReader([]byte("fallback"))), nil
+ }},
+ openRange: func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error) {
+ return nil, failure
+ },
+ }
+ rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
+ Location: g.claim.Claim.Holder.Location, Key: key, VersionID: "selected", Size: 8,
+ }}
+ rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 2, 3)
+ if rc != nil {
+ rc.Close()
+ }
+ want := failure
+ if errors.Is(failure, lease.ErrMirrorNotFound) {
+ want = os.ErrNotExist
+ }
+ if !errors.Is(err, want) || fullOpens != 0 {
+ t.Fatalf("range error = %v; want %v, full opens = %d", err, want, fullOpens)
+ }
+ })
+ }
+}
+
+func TestDoorPlanHTTPDeepRangeKeepsWatchdogAlive(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ const offset = 4 << 20
+ want := bytes.Repeat([]byte("tail"), 64<<10)
+ payload := append(bytes.Repeat([]byte("p"), offset), want...)
+ key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
+ id := model.Put("data-test", key, payload)
+ var read, fullOpens, rangeOpens atomic.Int64
+ fx.srv.dbStreamIdle = 250 * time.Millisecond
+ fx.srv.cfg.DBVersions = doorRangeHook{
+ doorOpenHook: doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
+ fullOpens.Add(1)
+ rc, err := model.OpenVersion(ctx, b, k, v)
+ return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
+ }},
+ openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
+ rangeOpens.Add(1)
+ if v != id || off != offset || length != int64(len(want)) {
+ return nil, fmt.Errorf("unexpected range %s %d %d", v, off, length)
+ }
+ rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
+ return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
+ },
+ }
+ status, _, body := rawReq(t, http.MethodGet, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member, "")
+ var list dbdoor.ListResponse
+ if err := json.Unmarshal([]byte(body), &list); err != nil || status != 200 || len(list.Files) != 1 {
+ t.Fatalf("list = %d %s, %v", status, body, err)
+ }
+ model.Put("data-test", key, []byte("replacement"))
+ ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+ defer cancel()
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header.Set("Authorization", "Bearer "+fx.member)
+ req.Header.Set(dbdoor.HandleHeader, list.Files[0].Handle)
+ req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
+ start := time.Now()
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := io.ReadAll(res.Body)
+ res.Body.Close()
+ if err != nil || res.StatusCode != 206 || !bytes.Equal(got, want) {
+ t.Fatalf("range = status %d, %d bytes, %v; want 206 and %d bytes", res.StatusCode, len(got), err, len(want))
+ }
+ if time.Since(start) <= 2*fx.srv.dbStreamIdle {
+ t.Fatal("stream did not outlive two watchdog intervals")
+ }
+ if read.Load() != int64(len(want)) || fullOpens.Load() != 0 || rangeOpens.Load() != 1 {
+ t.Fatalf("read %d bytes, full opens %d, range opens %d", read.Load(), fullOpens.Load(), rangeOpens.Load())
+ }
+ if res.Header.Get("Content-Range") != fmt.Sprintf("bytes %d-%d/%d", offset, len(payload)-1, len(payload)) || res.Header.Get(dbdoor.HandleHeader) != list.Files[0].Handle {
+ t.Fatalf("range headers = %v", res.Header)
+ }
+ fx.waitIdle(t, time.Second)
+}
diff --git a/internal/lease/leasetest/bucket.go b/internal/lease/leasetest/bucket.go
index fe13f4d98..94450f5e0 100644
--- a/internal/lease/leasetest/bucket.go
+++ b/internal/lease/leasetest/bucket.go
@@ -254,6 +254,31 @@ func (f *VersionedBucket) OpenVersion(ctx context.Context, bucket, key, versionI
return io.NopCloser(bytes.NewReader(b)), nil
}
+func versionRange(body []byte, offset, length int64) ([]byte, error) {
+ if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) || offset >= int64(len(body)) {
+ return nil, fmt.Errorf("leasetest: invalid version range")
+ }
+ n := int64(len(body)) - offset
+ if length > 0 && length < n {
+ n = length
+ }
+ return body[offset : offset+n], nil
+}
+
+func (f *VersionedBucket) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ b, err := f.read(bucket, key, versionID)
+ if err != nil {
+ return nil, err
+ }
+ b, err = versionRange(b, offset, length)
+ if err != nil {
+ return nil, err
+ }
+ return io.NopCloser(bytes.NewReader(b)), nil
+}
+
func (f *VersionedBucket) copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
if srcVersionID == "" {
return "", "", lease.ErrNoVersionID
@@ -338,3 +363,5 @@ func (f *VersionedBucket) PutIfAbsent(ctx context.Context, bucket, key string, b
}
var _ lease.VersionedStore = (*VersionedBucket)(nil)
+
+var _ lease.VersionRangeOpener = (*VersionedBucket)(nil)
diff --git a/internal/lease/leasetest/contract.go b/internal/lease/leasetest/contract.go
index d1a823408..235d75859 100644
--- a/internal/lease/leasetest/contract.go
+++ b/internal/lease/leasetest/contract.go
@@ -105,6 +105,91 @@ func RunVersionedStoreContract(t *testing.T, mk Maker) {
}
})
+ t.Run("range by exact version", func(t *testing.T) {
+ st, m, b, served := mk(t)
+ ranged, ok := st.(lease.VersionRangeOpener)
+ if !ok {
+ t.Skip("store does not support ranged version reads")
+ }
+ key := "dir/a b+c%?d"
+ id := m.Put(b, key, []byte("original"))
+ m.Put(b, key, []byte("replacement"))
+ marker := m.BareDelete(b, key)
+ for _, tc := range []struct {
+ name string
+ offset, length int64
+ want string
+ }{
+ {"bounded", 2, 3, "igi"},
+ {"to end", 3, -1, "ginal"},
+ {"past end", 6, 100, "al"},
+ {"whole", 0, -1, "original"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ rc, err := ranged.OpenVersionRange(ctx, b, key, id, tc.offset, tc.length)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ closeErr := rc.Close()
+ if err != nil || closeErr != nil || string(body) != tc.want {
+ t.Fatalf("range = %q, %v, close %v; want %q", body, err, closeErr, tc.want)
+ }
+ })
+ }
+ for _, tc := range []struct {
+ name, bucket, version string
+ offset, length int64
+ want error
+ local bool
+ }{
+ {"missing version ID", b, "", 0, 1, lease.ErrNoVersionID, true},
+ {"negative offset", b, id, -1, 1, nil, true},
+ {"zero length", b, id, 0, 0, nil, true},
+ {"overflow", b, id, 2, 1<<63 - 1, nil, true},
+ {"unsatisfiable", b, id, 8, -1, nil, false},
+ {"missing version", b, "missing", 0, 1, lease.ErrMirrorNotFound, false},
+ {"marker", b, marker, 0, 1, lease.ErrIsDeleteMarker, false},
+ {"missing bucket", "nobucket", id, 0, 1, nil, false},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ before := 0
+ if served != nil {
+ before = served()
+ }
+ rc, err := ranged.OpenVersionRange(ctx, tc.bucket, key, tc.version, tc.offset, tc.length)
+ if rc != nil {
+ rc.Close()
+ }
+ if err == nil || (tc.want != nil && !errors.Is(err, tc.want)) {
+ t.Fatalf("range error = %v; want %v", err, tc.want)
+ }
+ if tc.want == nil && errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("range error was treated as missing version: %v", err)
+ }
+ if tc.local && served != nil && served() != before {
+ t.Fatal("invalid range sent a request")
+ }
+ })
+ }
+ if err := st.DeleteVersion(ctx, b, key, id); err != nil {
+ t.Fatal(err)
+ }
+ if rc, err := ranged.OpenVersionRange(ctx, b, key, id, 2, 3); !errors.Is(err, lease.ErrMirrorNotFound) {
+ if rc != nil {
+ rc.Close()
+ }
+ t.Fatalf("deleted version range = %v", err)
+ }
+ m.FailOn = func(op, key string) error { return errors.New("offline") }
+ if rc, err := ranged.OpenVersionRange(ctx, b, key, id, 2, 3); err == nil || errors.Is(err, lease.ErrMirrorNotFound) {
+ if rc != nil {
+ rc.Close()
+ }
+ t.Fatalf("failed range = %v", err)
+ }
+ })
+
t.Run("copy by exact version; create-only; markers", func(t *testing.T) {
st, m, b, _ := mk(t)
k := "src/a b+c%?d"
diff --git a/internal/lease/leasetest/s3server.go b/internal/lease/leasetest/s3server.go
index f0877aeb5..92cba0b9d 100644
--- a/internal/lease/leasetest/s3server.go
+++ b/internal/lease/leasetest/s3server.go
@@ -3,9 +3,11 @@ package leasetest
import (
"encoding/xml"
"errors"
+ "fmt"
"io"
"net/http"
"net/url"
+ "strconv"
"strings"
"sync/atomic"
"time"
@@ -110,6 +112,29 @@ func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int
writeErr(w, err)
return
}
+ if hdr := r.Header.Get("Range"); hdr != "" {
+ span, ok := strings.CutPrefix(hdr, "bytes=")
+ start, end, dash := strings.Cut(span, "-")
+ offset, startErr := strconv.ParseInt(start, 10, 64)
+ length := int64(-1)
+ if end != "" {
+ last, endErr := strconv.ParseInt(end, 10, 64)
+ if endErr != nil || last < offset || last-offset == 1<<63-1 {
+ s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange")
+ return
+ }
+ length = last - offset + 1
+ }
+ total := len(body)
+ body, err = versionRange(body, offset, length)
+ if !ok || !dash || startErr != nil || err != nil {
+ s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange")
+ return
+ }
+ w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+int64(len(body))-1, total))
+ w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+ w.WriteHeader(http.StatusPartialContent)
+ }
w.Write(body)
case r.Method == http.MethodPut && r.Header.Get("x-amz-copy-source") != "":
f.serveCopy(w, r, bucket, key)
diff --git a/internal/lease/versioned.go b/internal/lease/versioned.go
index 54377d71b..c07bd037e 100644
--- a/internal/lease/versioned.go
+++ b/internal/lease/versioned.go
@@ -63,6 +63,13 @@ type VersionedStore interface {
PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (versionID string, err error)
}
+// VersionRangeOpener optionally streams a range of one exact object version.
+// A negative length reads to the end; a positive length bounds the read.
+// Zero length and negative offsets are invalid. Errors match OpenVersion.
+type VersionRangeOpener interface {
+ OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error)
+}
+
// versionedS3 binds every bucket to its own regional client explicitly.
// An unbound bucket is an error, never a guess.
type versionedS3 struct {
@@ -164,6 +171,24 @@ func mapReadErr(err error) error {
}
func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ return s.openVersion(ctx, bucket, key, versionID, nil)
+}
+
+func (s *versionedS3) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) {
+ if versionID == "" {
+ return nil, ErrNoVersionID
+ }
+ if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) {
+ return nil, errors.New("lease: invalid version range")
+ }
+ rangeHeader := fmt.Sprintf("bytes=%d-", offset)
+ if length > 0 {
+ rangeHeader += fmt.Sprint(offset + length - 1)
+ }
+ return s.openVersion(ctx, bucket, key, versionID, &rangeHeader)
+}
+
+func (s *versionedS3) openVersion(ctx context.Context, bucket, key, versionID string, byteRange *string) (io.ReadCloser, error) {
if versionID == "" {
return nil, ErrNoVersionID
}
@@ -171,7 +196,7 @@ func (s *versionedS3) OpenVersion(ctx context.Context, bucket, key, versionID st
if err != nil {
return nil, err
}
- out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID)})
+ out, err := c.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(versionID), Range: byteRange})
if err != nil {
if m := mapReadErr(err); m != err {
return nil, m
@@ -266,3 +291,5 @@ func (s *versionedS3) PutIfAbsent(ctx context.Context, bucket, key string, body
}
var _ VersionedStore = (*versionedS3)(nil)
+
+var _ VersionRangeOpener = (*versionedS3)(nil)
diff --git a/internal/lease/versioned_test.go b/internal/lease/versioned_test.go
index a7d415af3..2d7ad8f1f 100644
--- a/internal/lease/versioned_test.go
+++ b/internal/lease/versioned_test.go
@@ -2,6 +2,8 @@ package lease_test
import (
"context"
+ "io"
+ "net/http"
"net/http/httptest"
"testing"
@@ -95,3 +97,52 @@ func TestVersionedS3CopyEscapesLikeAWS(t *testing.T) {
t.Fatalf("copied %q, want the pinned old version", cur)
}
}
+
+func TestVersionedS3RangeWire(t *testing.T) {
+ m := leasetest.NewVersionedBucket()
+ m.AddBucket("data")
+ key := "dir/a b+c%?d"
+ id := m.Put("data", key, []byte("original"))
+ m.Put("data", key, []byte("replacement"))
+ handler, _ := leasetest.NewS3Server(m, "data")
+ for _, tc := range []struct {
+ length int64
+ header, contentRange, body string
+ }{
+ {3, "bytes=2-4", "bytes 2-4/8", "igi"},
+ {-1, "bytes=2-", "bytes 2-7/8", "iginal"},
+ } {
+ t.Run(tc.header, func(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodGet || r.URL.Query().Get("versionId") != id || r.Header.Get("Range") != tc.header {
+ t.Errorf("request = %s %s, Range %q", r.Method, r.URL, r.Header.Get("Range"))
+ }
+ rec := httptest.NewRecorder()
+ handler.ServeHTTP(rec, r)
+ if rec.Code != http.StatusPartialContent || rec.Header().Get("Content-Range") != tc.contentRange || rec.Body.String() != tc.body {
+ t.Errorf("response = %d %v %q", rec.Code, rec.Header(), rec.Body.String())
+ }
+ for k, values := range rec.Header() {
+ w.Header()[k] = values
+ }
+ w.WriteHeader(rec.Code)
+ w.Write(rec.Body.Bytes())
+ }))
+ defer srv.Close()
+ st := lease.NewVersionedS3WithClients(map[string]*awss3.Client{"data": wireClient(t, srv.URL)})
+ ranged, ok := st.(lease.VersionRangeOpener)
+ if !ok {
+ t.Fatal("S3 store lacks ranged version reads")
+ }
+ rc, err := ranged.OpenVersionRange(t.Context(), "data", key, id, 2, tc.length)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil || string(body) != tc.body {
+ t.Fatalf("read = %q, %v", body, err)
+ }
+ })
+ }
+}