chore: upgrade synapse-go to v0.3.2

Author: strahe

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/knadh/koanf/v2 v2.3.4
 	github.com/multiformats/go-multihash v0.2.3
 	github.com/prometheus/client_golang v1.23.2
-	github.com/strahe/synapse-go v0.3.1
+	github.com/strahe/synapse-go v0.3.2
 	github.com/uptrace/bun v1.2.18
 	github.com/uptrace/bun/dialect/pgdialect v1.2.18
 	github.com/uptrace/bun/dialect/sqlitedialect v1.2.18

go.sum

@@ -390,8 +390,8 @@ github.com/smira/go-statsd v1.3.4 h1:kBYWcLSGT+qC6JVbvfz48kX7mQys32fjDOPrfmsSx2c
 github.com/smira/go-statsd v1.3.4/go.mod h1:RjdsESPgDODtg1VpVVf9MJrEW2Hw0wtRNbmB1CAhu6A=
 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
-github.com/strahe/synapse-go v0.3.1 h1:GR+CIRGDxh9LRAXDF35NJTkUUXJap2HdOb5aML56G+U=
-github.com/strahe/synapse-go v0.3.1/go.mod h1:8aoCafxC0tCz4Ow7fodZipbsbVojVNiUh6nBxLodxjs=
+github.com/strahe/synapse-go v0.3.2 h1:dRnCTPdWoJTGhK0pWQoyDYV56Mvux9vD9bB+iYqYdLg=
+github.com/strahe/synapse-go v0.3.2/go.mod h1:8aoCafxC0tCz4Ow7fodZipbsbVojVNiUh6nBxLodxjs=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=

internal/admin/api_overview.go

@@ -103,22 +103,16 @@ func (s *Server) handleAPIOverview(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Objects
-	objCounts, err := s.repos.Objects.CountByState(ctx)
+	objCounts, err := s.repos.Objects.AggregateByState(ctx)
 	if err != nil {
-		s.logger.Warn("overview: failed to count objects", "error", err)
+		s.logger.Warn("overview: failed to aggregate objects", "error", err)
 	} else {
 		for _, oc := range objCounts {
 			resp.Objects.ByState[oc.State] = oc.Count
 			resp.Objects.Total += oc.Count
+			resp.Objects.TotalSizeBytes += oc.TotalSize
 		}
 	}
-
-	totalSize, err := s.repos.Objects.TotalSize(ctx)
-	if err != nil {
-		s.logger.Warn("overview: failed to get total size", "error", err)
-	} else {
-		resp.Objects.TotalSizeBytes = totalSize
-	}
 	objectAttention, err := s.repos.Objects.CountOverviewAttention(ctx)
 	if err != nil {
 		s.logger.Warn("overview: failed to count object attention", "error", err)

internal/admin/api_overview_test.go

@@ -62,8 +62,9 @@ func TestAPIOverviewIncludesAttentionAndActivePipeline(t *testing.T) {
 
 	var body struct {
 		Objects struct {
-			ByState   map[string]int64 `json:"by_state"`
-			Attention struct {
+			ByState        map[string]int64 `json:"by_state"`
+			TotalSizeBytes int64            `json:"total_size_bytes"`
+			Attention      struct {
 				NeedsAttention int64 `json:"needs_attention"`
 				Unavailable    int64 `json:"unavailable"`
 			} `json:"attention"`
@@ -87,6 +88,9 @@ func TestAPIOverviewIncludesAttentionAndActivePipeline(t *testing.T) {
 	if body.Objects.ByState[string(model.ObjectStateCached)] == 0 {
 		t.Fatal("overview should keep legacy object state counts")
 	}
+	if body.Objects.TotalSizeBytes != 30 {
+		t.Fatalf("total_size_bytes = %d, want 30", body.Objects.TotalSizeBytes)
+	}
 	if body.Tasks.ByStatus[string(model.TaskStatusCompleted)] == 0 {
 		t.Fatal("overview should keep legacy task status counts")
 	}

internal/db/repository/interfaces.go

@@ -157,9 +157,9 @@ type ObjectRepository interface {
 	ResetStaleVersionStates(ctx context.Context, fromState, toState model.ObjectState, staleBefore time.Time) (int, error)
 	// CountByState returns object counts grouped by state.
 	CountByState(ctx context.Context) ([]ObjectStateCount, error)
+	// AggregateByState returns object counts and sizes grouped by state.
+	AggregateByState(ctx context.Context) ([]ObjectStateAggregate, error)
 	CountOverviewAttention(ctx context.Context) (ObjectAttentionCount, error)
-	// TotalSize returns the sum of current object sizes in bytes.
-	TotalSize(ctx context.Context) (int64, error)
 	// CountByBucket returns the number of current objects in a bucket.
 	CountByBucket(ctx context.Context, bucketID int64) (int64, error)
 	// TotalSizeByBucket returns the sum of current object sizes in a bucket.
@@ -516,6 +516,13 @@ type ObjectStateCount struct {
 	Count int64  `bun:"count"`
 }
 
+// ObjectStateAggregate holds object count and total size grouped by state.
+type ObjectStateAggregate struct {
+	State     string `bun:"state"`
+	Count     int64  `bun:"count"`
+	TotalSize int64  `bun:"total_size"`
+}
+
 type ObjectAttentionCount struct {
 	NeedsAttention int64 `bun:"needs_attention"`
 	Unavailable    int64 `bun:"unavailable"`

internal/db/repository/object_repo.go

@@ -880,6 +880,22 @@ func (r *BunObjectRepo) CountByState(ctx context.Context) ([]ObjectStateCount, e
 	return counts, nil
 }
 
+// AggregateByState returns current object counts and sizes grouped by state.
+func (r *BunObjectRepo) AggregateByState(ctx context.Context) ([]ObjectStateAggregate, error) {
+	var rows []ObjectStateAggregate
+	err := r.db.NewSelect().
+		Model((*model.ObjectVersion)(nil)).
+		ColumnExpr("state, COUNT(*) AS count, COALESCE(SUM(size), 0) AS total_size").
+		Where("is_current = ?", true).
+		Where("is_delete_marker = ?", false).
+		GroupExpr("state").
+		Scan(ctx, &rows)
+	if err != nil {
+		return nil, fmt.Errorf("aggregating current object versions by state: %w", err)
+	}
+	return rows, nil
+}
+
 func (r *BunObjectRepo) CountOverviewAttention(ctx context.Context) (ObjectAttentionCount, error) {
 	var count ObjectAttentionCount
 	query := `WITH current_versions AS (
@@ -922,21 +938,6 @@ func (r *BunObjectRepo) CountOverviewAttention(ctx context.Context) (ObjectAtten
 	return count, nil
 }
 
-// TotalSize returns the sum of all current object sizes in bytes.
-func (r *BunObjectRepo) TotalSize(ctx context.Context) (int64, error) {
-	var total int64
-	err := r.db.NewSelect().
-		Model((*model.ObjectVersion)(nil)).
-		ColumnExpr("COALESCE(SUM(size), 0)").
-		Where("is_current = ?", true).
-		Where("is_delete_marker = ?", false).
-		Scan(ctx, &total)
-	if err != nil {
-		return 0, fmt.Errorf("computing total current object size: %w", err)
-	}
-	return total, nil
-}
-
 // CountByBucket returns the number of current objects in a bucket.
 func (r *BunObjectRepo) CountByBucket(ctx context.Context, bucketID int64) (int64, error) {
 	count, err := r.db.NewSelect().

internal/db/repository/object_repo_test.go

@@ -160,6 +160,43 @@ func TestObjectRepo_CountOverviewAttention(t *testing.T) {
 	}
 }
 
+func TestObjectRepo_AggregateByStateIncludesTotalSize(t *testing.T) {
+	db := testDB(t)
+	repos := repository.NewRepositories(db)
+	ctx := context.Background()
+	bucket := seedBucket(t, db, "state-aggregate-bucket")
+
+	cached := newObjectVersion(bucket.ID, "cached.txt", "01J00000000000000000000B01", 10)
+	if _, err := repos.Objects.CreateVersionAndSetCurrent(ctx, cached); err != nil {
+		t.Fatalf("seed cached version: %v", err)
+	}
+	failed := newObjectVersion(bucket.ID, "failed.txt", "01J00000000000000000000B02", 20)
+	if _, err := repos.Objects.CreateVersionAndSetCurrent(ctx, failed); err != nil {
+		t.Fatalf("seed failed version: %v", err)
+	}
+	if err := repos.Objects.UpdateVersionStateToFailed(ctx, failed.VersionID, model.ObjectStateCached, "upload failed"); err != nil {
+		t.Fatalf("mark failed version: %v", err)
+	}
+	if _, err := repos.Objects.CreateDeleteMarkerAndSetCurrent(ctx, bucket.ID, "deleted.txt", "01J00000000000000000000B03"); err != nil {
+		t.Fatalf("seed delete marker: %v", err)
+	}
+
+	rows, err := repos.Objects.AggregateByState(ctx)
+	if err != nil {
+		t.Fatalf("AggregateByState: %v", err)
+	}
+	byState := make(map[string]repository.ObjectStateAggregate, len(rows))
+	for _, row := range rows {
+		byState[row.State] = row
+	}
+	if got := byState[string(model.ObjectStateCached)]; got.Count != 1 || got.TotalSize != 10 {
+		t.Fatalf("cached aggregate = count:%d size:%d, want 1/10", got.Count, got.TotalSize)
+	}
+	if got := byState[string(model.ObjectStateFailed)]; got.Count != 1 || got.TotalSize != 20 {
+		t.Fatalf("failed aggregate = count:%d size:%d, want 1/20", got.Count, got.TotalSize)
+	}
+}
+
 func TestObjectRepo_CreateVersionAndSetCurrent_ConcurrentFirstUpload(t *testing.T) {
 	sqldb, err := sql.Open("sqlite", "file:"+filepath.Join(t.TempDir(), "objects.db")+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)")
 	if err != nil {

internal/securetoken/token.go

@@ -3,10 +3,14 @@ package securetoken
 import (
 	"crypto/rand"
 	"encoding/base64"
+	"fmt"
 )
 
 // URL generates a URL-safe random token from n bytes of entropy.
 func URL(n int) (string, error) {
+	if n <= 0 {
+		return "", fmt.Errorf("securetoken: entropy bytes must be positive")
+	}
 	buf := make([]byte, n)
 	if _, err := rand.Read(buf); err != nil {
 		return "", err

internal/securetoken/token_test.go

@@ -0,0 +1,47 @@
+package securetoken
+
+import (
+	"encoding/base64"
+	"strconv"
+	"testing"
+)
+
+func TestURLReturnsRawURLTokenWithRequestedEntropy(t *testing.T) {
+	for _, n := range []int{16, 20, 32} {
+		t.Run(strconv.Itoa(n), func(t *testing.T) {
+			token, err := URL(n)
+			if err != nil {
+				t.Fatalf("URL(%d) error = %v, want nil", n, err)
+			}
+			if token == "" {
+				t.Fatalf("URL(%d) token is empty", n)
+			}
+			decoded, err := base64.RawURLEncoding.DecodeString(token)
+			if err != nil {
+				t.Fatalf("URL(%d) token is not raw URL base64: %v", n, err)
+			}
+			if len(decoded) != n {
+				t.Fatalf("URL(%d) decoded bytes = %d, want %d", n, len(decoded), n)
+			}
+		})
+	}
+}
+
+func TestURLRejectsInvalidEntropy(t *testing.T) {
+	for _, n := range []int{0, -1} {
+		t.Run(strconv.Itoa(n), func(t *testing.T) {
+			defer func() {
+				if r := recover(); r != nil {
+					t.Fatalf("URL(%d) panicked, want error: %v", n, r)
+				}
+			}()
+			token, err := URL(n)
+			if err == nil {
+				t.Fatalf("URL(%d) error = nil, want error", n)
+			}
+			if token != "" {
+				t.Fatalf("URL(%d) token = %q, want empty", n, token)
+			}
+		})
+	}
+}

internal/worker/wallet_operation_runner_test.go

@@ -429,6 +429,7 @@ type fakeWalletOperator struct {
 	withdrawAmount *big.Int
 	onFund         func(context.Context)
 	blockFund      bool
+	fundErr        error
 }
 
 func (f *fakeWalletOperator) FundUSDFC(ctx context.Context, amount *big.Int) (string, error) {
@@ -440,6 +441,9 @@ func (f *fakeWalletOperator) FundUSDFC(ctx context.Context, amount *big.Int) (st
 		<-ctx.Done()
 		return "", ctx.Err()
 	}
+	if f.fundErr != nil {
+		return "", f.fundErr
+	}
 	return f.fundHash, nil
 }
 
@@ -515,3 +519,33 @@ func (f *fakeWalletEventPublisher) statuses() []model.WalletOperationStatus {
 	}
 	return statuses
 }
+
+func TestWalletOperationRunner_BroadcastErrorFailsOperation(t *testing.T) {
+	db := testutil.NewTestDB(t)
+	repos := repository.NewRepositories(db)
+	ctx := context.Background()
+	op, _, err := repos.WalletOperations.CreateOrGet(ctx, repository.CreateWalletOperationInput{
+		Type:            model.WalletOperationTypeFund,
+		ClientRequestID: "fund-broadcast-error",
+		Amount:          "100",
+	})
+	if err != nil {
+		t.Fatalf("CreateOrGet: %v", err)
+	}
+
+	operator := &fakeWalletOperator{fundErr: errors.New("rpc node offline")}
+	runner := NewWalletOperationRunner(repos, operator, nil, time.Millisecond, nil)
+
+	runner.runOnce(ctx)
+
+	got, err := repos.WalletOperations.GetByID(ctx, op.ID)
+	if err != nil {
+		t.Fatalf("GetByID: %v", err)
+	}
+	if got.Status != model.WalletOperationStatusFailed {
+		t.Fatalf("status = %q, want failed", got.Status)
+	}
+	if got.LastError == nil || !strings.Contains(*got.LastError, "rpc node offline") {
+		t.Fatalf("last_error = %v, want rpc node offline", got.LastError)
+	}
+}