amadan

carlos / platform Public

Bound each buffered storage read without capping whole sweeps

claude c2cc02e21966d4d792d8d07007f5259d820e8894 parent fb0c6df
5 files changed, +345 −7
  • internal/console/economicspass.go +4 −1
  • internal/console/economicspass_test.go +108 −0
  • internal/release/readtimeout.go +28 −0
  • internal/release/readtimeout_test.go +187 −0
  • internal/release/store.go +18 −6

CI passed — run details

diff --git a/internal/console/economicspass.go b/internal/console/economicspass.go
index 80076fb06..56aff686e 100644
--- a/internal/console/economicspass.go
+++ b/internal/console/economicspass.go
@@ -202,6 +202,7 @@ func (s *Server) storagePass(ctx context.Context, date string) error {
}
func (s *Server) collectStorageDay(ctx context.Context, date string) (*usage.StorageDay, error) {
+ ctx = release.WithBufferedReadTimeout(ctx, storageOperationTimeout)
key := usage.StorageDayKey(date)
if _, _, err := s.cfg.Releases.Get(ctx, key); err == nil {
return nil, nil // today already written (by us or a sibling console)
@@ -2002,10 +2003,12 @@ func sizeProvisionedStore(ctx context.Context, p *objstore.Provisioned) (int64,
var total int64
var token *string
for {
- out, err := client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
+ readCtx, cancel := release.BufferedReadContext(ctx)
+ out, err := client.ListObjectsV2(readCtx, &awss3.ListObjectsV2Input{
Bucket: aws.String(p.Bucket),
ContinuationToken: token,
})
+ cancel()
if err != nil {
return 0, err
}
diff --git a/internal/console/economicspass_test.go b/internal/console/economicspass_test.go
index 81c3b681f..831b3a655 100644
--- a/internal/console/economicspass_test.go
+++ b/internal/console/economicspass_test.go
@@ -7,6 +7,7 @@ import (
"fmt"
"math"
"net/http"
+ "net/http/httptest"
"net/url"
"os"
"path/filepath"
@@ -22,6 +23,81 @@ import (
"github.com/carlosframework/platform/internal/usage"
)
+func isolateAWSConfigForStorageTimeoutTest(t *testing.T) {
+ t.Helper()
+ t.Setenv("AWS_ACCESS_KEY_ID", "fixture")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "fixture")
+ t.Setenv("AWS_SESSION_TOKEN", "")
+ t.Setenv("AWS_PROFILE", "")
+ t.Setenv("AWS_WEB_IDENTITY_TOKEN_FILE", "")
+ t.Setenv("AWS_ROLE_ARN", "")
+ t.Setenv("AWS_CONTAINER_CREDENTIALS_FULL_URI", "")
+ t.Setenv("AWS_CONTAINER_CREDENTIALS_RELATIVE_URI", "")
+ t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
+ t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(t.TempDir(), "no-credentials"))
+ t.Setenv("AWS_CONFIG_FILE", filepath.Join(t.TempDir(), "no-config"))
+ t.Setenv("CARLOS_S3_ENDPOINT", "")
+}
+
+func TestSizeProvisionedStoreRefreshesBufferedDeadlinePerPage(t *testing.T) {
+ isolateAWSConfigForStorageTimeoutTest(t)
+ pages := 0
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ time.Sleep(400 * time.Millisecond)
+ pages++
+ w.Header().Set("Content-Type", "application/xml")
+ truncated := pages < 3
+ token := ""
+ if truncated {
+ token = fmt.Sprintf("<NextContinuationToken>page-%d</NextContinuationToken>", pages+1)
+ }
+ fmt.Fprintf(w, `<?xml version="1.0" encoding="UTF-8"?><ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>bucket</Name><IsTruncated>%t</IsTruncated>%s<Contents><Key>key-%d</Key><Size>%d</Size></Contents></ListBucketResult>`, truncated, token, pages, pages)
+ }))
+ t.Cleanup(srv.Close)
+ p := &objstore.Provisioned{Bucket: "bucket", Region: "us-east-1", Endpoint: srv.URL}
+ start := time.Now()
+ total, err := sizeProvisionedStore(release.WithBufferedReadTimeout(context.Background(), time.Second), p)
+ elapsed := time.Since(start)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if total != 6 {
+ t.Fatalf("total = %d, want 6", total)
+ }
+ if elapsed <= time.Second {
+ t.Fatalf("three-page precondition elapsed only %v", elapsed)
+ }
+}
+
+func TestSizeProvisionedStoreTimesOutAStalledLaterPage(t *testing.T) {
+ isolateAWSConfigForStorageTimeoutTest(t)
+ releasePage := make(chan struct{})
+ pages := 0
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ pages++
+ if pages == 1 {
+ w.Header().Set("Content-Type", "application/xml")
+ fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?><ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>bucket</Name><IsTruncated>true</IsTruncated><NextContinuationToken>page-2</NextContinuationToken><Contents><Key>key-1</Key><Size>1</Size></Contents></ListBucketResult>`)
+ return
+ }
+ select {
+ case <-r.Context().Done():
+ case <-releasePage:
+ }
+ }))
+ t.Cleanup(srv.Close)
+ t.Cleanup(func() { close(releasePage) })
+ parent := context.Background()
+ p := &objstore.Provisioned{Bucket: "bucket", Region: "us-east-1", Endpoint: srv.URL}
+ _, err := sizeProvisionedStore(release.WithBufferedReadTimeout(parent, 30*time.Millisecond), p)
+ if !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("size error = %v", err)
+ }
+ if parent.Err() != nil {
+ t.Fatalf("parent error = %v", parent.Err())
+ }
+}
+
// putStoreJSON marshals v and writes it straight to the fixture bucket at
// key, create-only. Named distinctly from apiconfig_test.go's putJSON (an
// HTTP PUT helper against the running server) — same package, so a same-
@@ -65,6 +141,26 @@ type failAfterNStore struct {
calls int
}
+type bufferedReadObservingStore struct {
+ release.Store
+ getCalls int
+ firstGetMarked bool
+}
+
+func (s *bufferedReadObservingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ readCtx, cancel := release.BufferedReadContext(ctx)
+ defer cancel()
+ if s.getCalls == 0 {
+ _, s.firstGetMarked = readCtx.Deadline()
+ }
+ s.getCalls++
+ return s.Store.Get(ctx, key)
+}
+
+func (s *bufferedReadObservingStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) {
+ return s.Store.(release.SizeLister).ListSizes(ctx, prefix)
+}
+
func (s *failAfterNStore) Get(ctx context.Context, key string) ([]byte, string, error) {
if key == s.key {
s.calls++
@@ -85,6 +181,18 @@ const (
fixtureApp = "kass"
)
+func TestCollectStorageDayMarksItsFirstRead(t *testing.T) {
+ fx := newConsoleFixture(t)
+ observed := &bufferedReadObservingStore{Store: fx.rel}
+ fx.srv.cfg.Releases = observed
+ if _, err := fx.srv.collectStorageDay(context.Background(), "2026-10-03"); err != nil {
+ t.Fatal(err)
+ }
+ if !observed.firstGetMarked {
+ t.Fatal("first storage collection Get had no buffered-read deadline")
+ }
+}
+
func TestEconomicsPassWritesCensusOnce(t *testing.T) {
fx := newConsoleFixture(t)
ctx := context.Background()
diff --git a/internal/release/readtimeout.go b/internal/release/readtimeout.go
new file mode 100644
index 000000000..efc5d1096
--- /dev/null
+++ b/internal/release/readtimeout.go
@@ -0,0 +1,28 @@
+package release
+
+import (
+ "context"
+ "time"
+)
+
+type bufferedReadTimeoutKey struct{}
+
+// WithBufferedReadTimeout marks ctx so buffered provider reads use at most d
+// for one SDK operation, including the SDK's retries and backoff. Nested
+// markers retain the tighter duration. Streaming reads and writes ignore it.
+func WithBufferedReadTimeout(ctx context.Context, d time.Duration) context.Context {
+ if existing, ok := ctx.Value(bufferedReadTimeoutKey{}).(time.Duration); ok && existing < d {
+ d = existing
+ }
+ return context.WithValue(ctx, bufferedReadTimeoutKey{}, d)
+}
+
+// BufferedReadContext returns a fresh deadline for one buffered SDK
+// operation. Without a marker it returns ctx unchanged and a no-op cancel.
+func BufferedReadContext(ctx context.Context) (context.Context, context.CancelFunc) {
+ d, ok := ctx.Value(bufferedReadTimeoutKey{}).(time.Duration)
+ if !ok {
+ return ctx, func() {}
+ }
+ return context.WithTimeout(ctx, d)
+}
diff --git a/internal/release/readtimeout_test.go b/internal/release/readtimeout_test.go
new file mode 100644
index 000000000..192a4018e
--- /dev/null
+++ b/internal/release/readtimeout_test.go
@@ -0,0 +1,187 @@
+package release
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "sync"
+ "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"
+)
+
+func TestBufferedReadContextIsOptInAndNestedTightens(t *testing.T) {
+ parent := context.Background()
+ unmarked, cancel := BufferedReadContext(parent)
+ cancel()
+ if unmarked != parent {
+ t.Fatal("unmarked context was replaced")
+ }
+ if _, ok := unmarked.Deadline(); ok {
+ t.Fatal("unmarked context acquired a deadline")
+ }
+
+ marked := WithBufferedReadTimeout(parent, time.Hour)
+ marked = WithBufferedReadTimeout(marked, time.Minute)
+ marked = WithBufferedReadTimeout(marked, 2*time.Minute)
+ child, cancel := BufferedReadContext(marked)
+ defer cancel()
+ deadline, ok := child.Deadline()
+ if !ok || time.Until(deadline) > time.Minute || time.Until(deadline) < 59*time.Second {
+ t.Fatalf("nested deadline = %v, ok=%v", deadline, ok)
+ }
+
+ immediate, cancel := BufferedReadContext(WithBufferedReadTimeout(parent, 0))
+ defer cancel()
+ select {
+ case <-immediate.Done():
+ default:
+ t.Fatal("zero timeout did not cancel immediately")
+ }
+}
+
+func TestBufferedReadContextKeepsStrongerParentDeadline(t *testing.T) {
+ parent, parentCancel := context.WithTimeout(context.Background(), time.Hour)
+ defer parentCancel()
+ want, _ := parent.Deadline()
+ child, cancel := BufferedReadContext(WithBufferedReadTimeout(parent, 2*time.Hour))
+ defer cancel()
+ got, ok := child.Deadline()
+ if !ok || !got.Equal(want) {
+ t.Fatalf("deadline = %v, ok=%v; want parent %v", got, ok, want)
+ }
+}
+
+type recordingTransport struct {
+ base http.RoundTripper
+ mu sync.Mutex
+ ctxs []context.Context
+}
+
+func (r *recordingTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+ r.mu.Lock()
+ r.ctxs = append(r.ctxs, req.Context())
+ r.mu.Unlock()
+ return r.base.RoundTrip(req)
+}
+
+func (r *recordingTransport) contexts() []context.Context {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return append([]context.Context(nil), r.ctxs...)
+}
+
+func timeoutS3Store(serverURL string, client *http.Client) *S3Store {
+ return &S3Store{Bucket: "bucket", Client: awss3.New(awss3.Options{
+ Region: "us-east-1", BaseEndpoint: aws.String(serverURL), UsePathStyle: true,
+ Credentials: credentials.NewStaticCredentialsProvider("fixture", "fixture", ""),
+ RetryMaxAttempts: 1, HTTPClient: client,
+ })}
+}
+
+func TestS3BufferedReadsTimeoutAtHeadersAndDuringBody(t *testing.T) {
+ t.Run("bounded list waits for headers", func(t *testing.T) {
+ release := make(chan struct{})
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ select {
+ case <-r.Context().Done():
+ case <-release:
+ }
+ }))
+ t.Cleanup(srv.Close)
+ t.Cleanup(func() { close(release) })
+ parent := context.Background()
+ _, err := timeoutS3Store(srv.URL, srv.Client()).ListPage(WithBufferedReadTimeout(parent, 30*time.Millisecond), "apps/", "", 1)
+ if !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("ListPage error = %v", err)
+ }
+ if parent.Err() != nil {
+ t.Fatalf("parent error = %v", parent.Err())
+ }
+ })
+
+ t.Run("get body", func(t *testing.T) {
+ release := make(chan struct{})
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Length", "10")
+ _, _ = w.Write([]byte("x"))
+ w.(http.Flusher).Flush()
+ select {
+ case <-r.Context().Done():
+ case <-release:
+ }
+ }))
+ t.Cleanup(srv.Close)
+ t.Cleanup(func() { close(release) })
+ parent := context.Background()
+ _, _, err := timeoutS3Store(srv.URL, srv.Client()).Get(WithBufferedReadTimeout(parent, 30*time.Millisecond), "object")
+ if !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("Get error = %v", err)
+ }
+ if parent.Err() != nil {
+ t.Fatalf("parent error = %v", parent.Err())
+ }
+ })
+}
+
+func TestS3BufferedListingsRefreshDeadlineForEveryPage(t *testing.T) {
+ tests := []struct {
+ name string
+ call func(context.Context, *S3Store) error
+ }{
+ {"ListMod", func(ctx context.Context, s *S3Store) error { _, err := s.ListMod(ctx, "apps/"); return err }},
+ {"List", func(ctx context.Context, s *S3Store) error { _, err := s.List(ctx, "apps/"); return err }},
+ {"ListSizes", func(ctx context.Context, s *S3Store) error { _, err := s.ListSizes(ctx, "apps/"); return err }},
+ {"ListDirs", func(ctx context.Context, s *S3Store) error { _, err := s.ListDirs(ctx, "apps/"); return err }},
+ {"ListPage unbounded", func(ctx context.Context, s *S3Store) error { _, err := s.ListPage(ctx, "apps/", "", 0); return err }},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var mu sync.Mutex
+ pages := 0
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ time.Sleep(400 * time.Millisecond)
+ mu.Lock()
+ pages++
+ page := pages
+ mu.Unlock()
+ w.Header().Set("Content-Type", "application/xml")
+ truncated := page < 3
+ token := ""
+ if truncated {
+ token = fmt.Sprintf("<NextContinuationToken>page-%d</NextContinuationToken>", page+1)
+ }
+ fmt.Fprintf(w, `<?xml version="1.0" encoding="UTF-8"?><ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>bucket</Name><IsTruncated>%t</IsTruncated>%s<Contents><Key>apps/key-%d</Key><Size>%d</Size><LastModified>2026-10-03T12:00:00Z</LastModified></Contents><CommonPrefixes><Prefix>apps/dir-%d/</Prefix></CommonPrefixes></ListBucketResult>`, truncated, token, page, page, page)
+ }))
+ t.Cleanup(srv.Close)
+ recorder := &recordingTransport{base: srv.Client().Transport}
+ client := *srv.Client()
+ client.Transport = recorder
+ start := time.Now()
+ err := tt.call(WithBufferedReadTimeout(context.Background(), time.Second), timeoutS3Store(srv.URL, &client))
+ elapsed := time.Since(start)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if elapsed <= time.Second {
+ t.Fatalf("three-page precondition elapsed only %v", elapsed)
+ }
+ ctxs := recorder.contexts()
+ if len(ctxs) != 3 {
+ t.Fatalf("request contexts = %d, want 3", len(ctxs))
+ }
+ for i, ctx := range ctxs {
+ select {
+ case <-ctx.Done():
+ default:
+ t.Fatalf("page %d child context was not canceled", i+1)
+ }
+ }
+ })
+ }
+}
diff --git a/internal/release/store.go b/internal/release/store.go
index cd45453c5..6f1d168c9 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -595,7 +595,9 @@ func s3CASConflict(err error) bool {
}
func (s *S3Store) Get(ctx context.Context, key string) ([]byte, string, error) {
- out, err := s.Client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key})
+ readCtx, cancel := BufferedReadContext(ctx)
+ defer cancel()
+ out, err := s.Client.GetObject(readCtx, &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key})
if err != nil {
if s3NotFound(err) {
return nil, "", fmt.Errorf("s3store: %s: %w", key, os.ErrNotExist)
@@ -687,7 +689,9 @@ func (s *S3Store) ListPage(ctx context.Context, prefix, startAfter string, limit
var keys []string
p := awss3.NewListObjectsV2Paginator(s.Client, in)
for p.HasMorePages() {
- page, err := p.NextPage(ctx)
+ readCtx, cancel := BufferedReadContext(ctx)
+ page, err := p.NextPage(readCtx)
+ cancel()
if err != nil {
return nil, err
}
@@ -699,7 +703,9 @@ func (s *S3Store) ListPage(ctx context.Context, prefix, startAfter string, limit
}
n := int32(limit)
in.MaxKeys = &n
- out, err := s.Client.ListObjectsV2(ctx, in)
+ readCtx, cancel := BufferedReadContext(ctx)
+ out, err := s.Client.ListObjectsV2(readCtx, in)
+ cancel()
if err != nil {
return nil, err
}
@@ -719,7 +725,9 @@ func (s *S3Store) ListMod(ctx context.Context, prefix string) ([]KeyMod, error)
Prefix: &prefix,
})
for p.HasMorePages() {
- page, err := p.NextPage(ctx)
+ readCtx, cancel := BufferedReadContext(ctx)
+ page, err := p.NextPage(readCtx)
+ cancel()
if err != nil {
return nil, err
}
@@ -739,7 +747,9 @@ func (s *S3Store) ListSizes(ctx context.Context, prefix string) ([]KeySize, erro
Prefix: &prefix,
})
for p.HasMorePages() {
- page, err := p.NextPage(ctx)
+ readCtx, cancel := BufferedReadContext(ctx)
+ page, err := p.NextPage(readCtx)
+ cancel()
if err != nil {
return nil, err
}
@@ -765,7 +775,9 @@ func (s *S3Store) ListDirs(ctx context.Context, prefix string) ([]string, error)
Delimiter: aws.String("/"),
})
for p.HasMorePages() {
- page, err := p.NextPage(ctx)
+ readCtx, cancel := BufferedReadContext(ctx)
+ page, err := p.NextPage(readCtx)
+ cancel()
if err != nil {
return nil, err
}