amadan

carlos / platform Public

console: daily storage sweep with unswept degradation

storagePass sizes every app's deployment-bucket prefix via
release.SizeLister and, when an app has a provisioned tenant object store
(objstore.Record.Provisioned), sizes that too — through a storeSizer seam
on Server so tests can force a sizer failure without live AWS. A store the
sweep can't reach is marked StoreUnswept rather than reported as zero
bytes. Wired into economicsPass alongside censusPass so a census failure
still lets storage run.

sizeProvisionedStore (the real, non-test sizer) builds its S3 client via
the new objstore.S3ClientFor, factored out of NewAWSProvisioner so the
region/endpoint option layering (path-style + checksum quirks for
S3-compatible backends) has one definition instead of two.

deletableStore (view_test.go) gained a ListSizes forwarder, mirroring its
existing ListMod forwarder — its embedded field is the release.Store
interface, so ListSizes was otherwise silently dropped even though the
DirStore underneath answers it, making every console test bucket look
unable to size a real bucket.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell aaa157b6be68f1630b304f6eb035f781dfdd875d parent 95e0f7d
5 files changed, +241 −5
  • internal/console/economicspass.go +101 −1
  • internal/console/economicspass_test.go +102 −0
  • internal/console/server.go +5 −0
  • internal/console/view_test.go +12 −0
  • internal/objstore/aws.go +21 −4
diff --git a/internal/console/economicspass.go b/internal/console/economicspass.go
index edb34c080..fe09f3f0d 100644
--- a/internal/console/economicspass.go
+++ b/internal/console/economicspass.go
@@ -5,8 +5,13 @@ import (
"fmt"
"log"
"os"
+ "strings"
"time"
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+
+ "github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/usage"
)
@@ -44,7 +49,9 @@ func (s *Server) economicsPass(ctx context.Context, now time.Time) error {
if err := s.censusPass(ctx, date); err != nil {
log.Printf("console: census pass %s: %v", date, err)
}
- // Task 9 adds: s.storagePass(ctx, date)
+ if err := s.storagePass(ctx, date); err != nil {
+ log.Printf("console: storage pass %s: %v", date, err)
+ }
// Task 10 adds: s.statementsPass(ctx, now)
// Task 12 adds: s.billPass(ctx, now)
return nil
@@ -102,3 +109,96 @@ func (s *Server) censusPass(ctx context.Context, date string) error {
_, err = usage.CreateDayObject(ctx, s.cfg.Releases, key, day)
return err
}
+
+// storagePass sizes every app's deployment-bucket prefix, plus its tenant
+// object store when one is provisioned, create-only per day. Requires the
+// bucket to support sized listings (release.SizeLister); a store that
+// cannot answer marks the day's sweep absent rather than lying with zeros.
+func (s *Server) storagePass(ctx context.Context, date string) error {
+ key := usage.StorageDayKey(date)
+ if _, _, err := s.cfg.Releases.Get(ctx, key); err == nil {
+ return nil // today already written (by us or a sibling console)
+ }
+ sizer, ok := s.cfg.Releases.(release.SizeLister)
+ if !ok {
+ log.Printf("console: storage pass: bucket store does not support sized listings; skipping")
+ return nil
+ }
+ // ListDirs (both DirStore and S3Store) returns each child as a FULL,
+ // slash-terminated prefix — "apps/bab/", not "bab" — so the next level's
+ // call takes that string straight back as its own prefix, and the bare
+ // name for map keys/AppPrefix is recovered by trimming both ends.
+ acctPrefixes, err := s.cfg.Releases.ListDirs(ctx, "apps/")
+ if err != nil {
+ return err
+ }
+ day := usage.StorageDay{V: 1, Date: date, Apps: map[string]usage.AppStorage{}}
+ for _, acctPrefix := range acctPrefixes {
+ acct := strings.TrimSuffix(strings.TrimPrefix(acctPrefix, "apps/"), "/")
+ appPrefixes, err := s.cfg.Releases.ListDirs(ctx, acctPrefix)
+ if err != nil {
+ return err
+ }
+ for _, appPrefix := range appPrefixes {
+ app := strings.TrimSuffix(strings.TrimPrefix(appPrefix, acctPrefix), "/")
+ sizes, err := sizer.ListSizes(ctx, release.AppPrefix(acct, app))
+ if err != nil {
+ return err
+ }
+ var st usage.AppStorage
+ for _, ks := range sizes {
+ st.BucketBytes += ks.Size
+ }
+ if rec, _, err := objstore.Get(ctx, s.cfg.Releases, acct, app); err == nil && rec != nil && rec.Provisioned != nil {
+ n, serr := s.sizeStore(ctx, rec.Provisioned)
+ if serr != nil {
+ log.Printf("console: storage pass: store %s/%s: %v", acct, app, serr)
+ st.StoreUnswept = true
+ } else {
+ st.StoreBytes = n
+ }
+ }
+ day.Apps[acct+"/"+app] = st
+ }
+ }
+ _, err = usage.CreateDayObject(ctx, s.cfg.Releases, key, day)
+ return err
+}
+
+// sizeStore routes through the test seam or the real S3 sizer.
+func (s *Server) sizeStore(ctx context.Context, p *objstore.Provisioned) (int64, error) {
+ if s.storeSizer != nil {
+ return s.storeSizer(ctx, p)
+ }
+ return sizeProvisionedStore(ctx, p)
+}
+
+// sizeProvisionedStore sums a tenant object store's live bytes with a
+// paginated ListObjectsV2, using an S3 client built exactly the way
+// objstore.NewAWSProvisioner builds its own (objstore.S3ClientFor: region +
+// optional endpoint override for S3-compatible backends).
+func sizeProvisionedStore(ctx context.Context, p *objstore.Provisioned) (int64, error) {
+ client, err := objstore.S3ClientFor(ctx, p.Region, p.Endpoint)
+ if err != nil {
+ return 0, err
+ }
+ var total int64
+ var token *string
+ for {
+ out, err := client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
+ Bucket: aws.String(p.Bucket),
+ ContinuationToken: token,
+ })
+ if err != nil {
+ return 0, err
+ }
+ for _, obj := range out.Contents {
+ total += aws.ToInt64(obj.Size)
+ }
+ if !aws.ToBool(out.IsTruncated) {
+ break
+ }
+ token = out.NextContinuationToken
+ }
+ return total, nil
+}
diff --git a/internal/console/economicspass_test.go b/internal/console/economicspass_test.go
index 8b60110f5..9d18c3b9d 100644
--- a/internal/console/economicspass_test.go
+++ b/internal/console/economicspass_test.go
@@ -3,15 +3,27 @@ package console
import (
"context"
"encoding/json"
+ "errors"
"os"
"path/filepath"
"testing"
"time"
+ "github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/usage"
)
+// fixtureAcct/fixtureApp are the account and app buildFixtures (view_test.go)
+// actually ships releases under: release.OpsAccount ("bab") owns both "kass"
+// and "otherapp" routes, but only "kass" gets shipped artifacts
+// (apps/bab/kass/releases/...) — "otherapp" has a registry route and no
+// bucket prefix at all, so it never appears in a storage sweep.
+const (
+ fixtureAcct = release.OpsAccount
+ fixtureApp = "kass"
+)
+
func TestEconomicsPassWritesCensusOnce(t *testing.T) {
fx := newConsoleFixture(t)
ctx := context.Background()
@@ -117,3 +129,93 @@ func TestStoreAccountsListsAll(t *testing.T) {
}
}
}
+
+func TestStoragePassSizesAppsAndDegradesStores(t *testing.T) {
+ fx := newConsoleFixture(t)
+ ctx := context.Background()
+
+ // The fixture bucket already has releases under apps/<acct>/<app>/ from
+ // buildFixtures; add a known-size object under one of those app prefixes
+ // and read its acct/app from the fixture rather than hardcoding.
+ if _, err := fx.rel.Put(ctx, "apps/"+fixtureAcct+"/"+fixtureApp+"/releases/x.tgz", make([]byte, 2048), ""); err != nil {
+ t.Fatal(err)
+ }
+
+ fx.srv.storeSizer = func(ctx context.Context, p *objstore.Provisioned) (int64, error) {
+ return 0, errors.New("endpoint down")
+ }
+
+ if err := fx.srv.storagePass(ctx, "2026-08-11"); err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := fx.rel.Get(ctx, usage.StorageDayKey("2026-08-11"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var day usage.StorageDay
+ if err := json.Unmarshal(raw, &day); err != nil {
+ t.Fatal(err)
+ }
+ app := day.Apps[fixtureAcct+"/"+fixtureApp]
+ if app.BucketBytes < 2048 {
+ t.Fatalf("bucket bytes = %d, want >= 2048", app.BucketBytes)
+ }
+ // buildFixtures provisions no object store for this app — the injected
+ // storeSizer (which always errors) must never even be consulted, so the
+ // app must not come back marked unswept.
+ if app.StoreUnswept {
+ t.Fatal("app with no provisioned store came back StoreUnswept")
+ }
+
+ // Idempotence: a second run the same day must not error or rewrite.
+ before := string(raw)
+ if err := fx.srv.storagePass(ctx, "2026-08-11"); err != nil {
+ t.Fatal(err)
+ }
+ after, _, _ := fx.rel.Get(ctx, usage.StorageDayKey("2026-08-11"))
+ if string(after) != before {
+ t.Fatal("second storage pass rewrote the day")
+ }
+}
+
+// TestStoragePassMarksUnreachableStoreUnswept: buildFixtures seeds no
+// provisioned object store, so this test writes a minimal objstore.Record
+// with a Provisioned block directly and confirms a sizer failure marks the
+// day's entry StoreUnswept rather than lying with a zero byte count.
+func TestStoragePassMarksUnreachableStoreUnswept(t *testing.T) {
+ fx := newConsoleFixture(t)
+ ctx := context.Background()
+
+ rec := &objstore.Record{
+ Provisioned: &objstore.Provisioned{
+ Bucket: "test-tenant-bucket",
+ Region: "us-east-1",
+ },
+ }
+ if _, err := objstore.Put(ctx, fx.rel, fixtureAcct, fixtureApp, rec, ""); err != nil {
+ t.Fatal(err)
+ }
+
+ fx.srv.storeSizer = func(ctx context.Context, p *objstore.Provisioned) (int64, error) {
+ return 0, errors.New("endpoint down")
+ }
+
+ if err := fx.srv.storagePass(ctx, "2026-08-11"); err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := fx.rel.Get(ctx, usage.StorageDayKey("2026-08-11"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var day usage.StorageDay
+ if err := json.Unmarshal(raw, &day); err != nil {
+ t.Fatal(err)
+ }
+ app := day.Apps[fixtureAcct+"/"+fixtureApp]
+ if !app.StoreUnswept {
+ t.Fatalf("app with unreachable store = %+v, want StoreUnswept", app)
+ }
+ if app.StoreBytes != 0 {
+ t.Fatalf("app.StoreBytes = %d, want 0 when unswept", app.StoreBytes)
+ }
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index fada08534..6eb745a45 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -268,6 +268,11 @@ type Server struct {
// dashboard and every owner's app-metrics page don't each re-sweep 48h of
// chunks per request.
metricsCache metricsCache
+ // storeSizer is the storage pass's test seam for sizing a provisioned
+ // tenant object store: nil means "use the real S3 sizer"
+ // (sizeProvisionedStore, economicspass.go); tests inject a fake so a
+ // sweep can be proven to degrade ("unswept") without live AWS.
+ storeSizer func(ctx context.Context, p *objstore.Provisioned) (int64, error)
// shipSem bounds how many `carlos ship` uploads (apicli.go,
// handleAPIShip) may run AT ONCE, as opposed to s.cli.ship, which
// bounds them over TIME. A rate limiter is a counter; it says nothing
diff --git a/internal/console/view_test.go b/internal/console/view_test.go
index 691e5e1d6..81d5bfd9f 100644
--- a/internal/console/view_test.go
+++ b/internal/console/view_test.go
@@ -60,6 +60,18 @@ func (d *deletableStore) ListMod(ctx context.Context, prefix string) ([]release.
return ml.ListMod(ctx, prefix)
}
+// ListSizes forwards to the wrapped store's SizeLister for the same reason
+// ListMod does above: the economics storage sweep (console.storagePass) is
+// a real consumer, and a double that silently drops what the concrete store
+// underneath can answer makes the sweep look unable to size a real bucket.
+func (d *deletableStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) {
+ sl, ok := d.Store.(release.SizeLister)
+ if !ok {
+ return nil, fmt.Errorf("deletablestore: wrapped store has no ListSizes")
+ }
+ return sl.ListSizes(ctx, prefix)
+}
+
// buildFixtures makes a little platform: a registry with routes for apps
// kass (ready instance on stable) and otherapp, and a deployment bucket with
// kass v1 promoted to stable and v2 to canary/x. Shared with server_test.go.
diff --git a/internal/objstore/aws.go b/internal/objstore/aws.go
index 02cec85f7..bfaaa5993 100644
--- a/internal/objstore/aws.go
+++ b/internal/objstore/aws.go
@@ -74,20 +74,37 @@ var _ Provisioner = (*awsProvisioner)(nil)
// twin, so a non-AWS backend (endpoint set, "aws" backend) still means
// IAM-against-real-AWS in v1 — there is nothing else for it to point at.
func NewAWSProvisioner(ctx context.Context, region, endpoint string) (Provisioner, error) {
+ s3Client, err := S3ClientFor(ctx, region, endpoint)
+ if err != nil {
+ return nil, err
+ }
cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
if err != nil {
return nil, fmt.Errorf("objstore: load aws config: %w", err)
}
- s3Client := awss3.NewFromConfig(cfg, func(o *awss3.Options) {
+ iamClient := iam.NewFromConfig(cfg)
+ return &awsProvisioner{s3: s3Client, iam: iamClient, region: region}, nil
+}
+
+// S3ClientFor builds a bare *s3.Client the same way NewAWSProvisioner builds
+// its own — region + the s3compat-style option layering (BaseEndpoint,
+// UsePathStyle, the checksum pair) applied only when endpoint is non-empty —
+// for callers that need the raw client rather than the Provisioner seam.
+// The economics storage sweep (console.sizeProvisionedStore) is the first:
+// it sizes a tenant's provisioned bucket directly, with no IAM involved.
+func S3ClientFor(ctx context.Context, region, endpoint string) (*awss3.Client, error) {
+ cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
+ if err != nil {
+ return nil, fmt.Errorf("objstore: load aws config: %w", err)
+ }
+ return awss3.NewFromConfig(cfg, func(o *awss3.Options) {
if endpoint != "" {
o.BaseEndpoint = aws.String(endpoint)
o.UsePathStyle = true
o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
o.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
}
- })
- iamClient := iam.NewFromConfig(cfg)
- return &awsProvisioner{s3: s3Client, iam: iamClient, region: region}, nil
+ }), nil
}
// ---- policy document --------------------------------------------------