perf: measurable latency & efficiency wins across gofin services

docker-compose.yml

@@ -70,6 +70,10 @@ services:
       - EXPENSE_SERVICE_REST=http://expense-service:8082
       - FINANCE_SERVICE_REST=http://finance-service:8083
       - DATARIGHTS_SERVICE_REST=http://datarights-service:8084
+      # Upper bound on the auth ValidateToken gRPC call (parsed as a Go
+      # duration, default 3s). Caps tail latency if the auth service hangs;
+      # must comfortably exceed a healthy validation (low milliseconds).
+      - GATEWAY_VALIDATE_TIMEOUT=${GATEWAY_VALIDATE_TIMEOUT:-3s}
       - JWT_SECRET=${JWT_SECRET}
       - LOG_LEVEL=${LOG_LEVEL:-info}
       - ENVIRONMENT=${ENVIRONMENT:-development}

docs/architecture.md

@@ -247,7 +247,7 @@ sequenceDiagram
 
 ## Data Flow: Data Export
 
-When a user requests a data export, the datarights service collects data from all upstream services asynchronously and delivers the result via email:
+When a user requests a data export, the datarights service collects data from all upstream services concurrently in a background goroutine (an `errgroup` fan-out, with ZIP assembly as the fan-in barrier) and delivers the result via email:
 
 ```mermaid
 sequenceDiagram
@@ -268,12 +268,17 @@ sequenceDiagram
 
     Note over DR: Async goroutine starts
     DR->>P: UPDATE status=running
-    DR->>A: gRPC: GetUser (profile + email)
-    A-->>DR: user profile
-    DR->>E: gRPC: GetAllUserExpenses (paginated)
-    E-->>DR: expense data
-    DR->>F: gRPC: GetAllUserData (tags, periods, defaults)
-    F-->>DR: finance data
+    par Providers collect concurrently (errgroup fan-out)
+        DR->>A: gRPC: GetUser (profile + email)
+        A-->>DR: user profile
+    and
+        DR->>E: gRPC: StreamAllUserExpenses (server-streaming)
+        E-->>DR: stream expense rows (Recv loop, written incrementally)
+    and
+        DR->>F: gRPC: GetAllUserData (tags, periods, defaults)
+        F-->>DR: finance data
+    end
+    Note over DR: Fan-in barrier: all providers complete → assemble ZIP
     DR->>DR: Generate CSVs + ZIP
     DR->>R: Send email with ZIP attachment
     R-->>DR: delivery confirmed

docs/data-export.md

@@ -114,7 +114,7 @@ Single row with current budget defaults (empty if not configured).
 
 ## Adding a Data Provider
 
-The export engine uses a provider registry pattern. Each provider is responsible for one CSV file.
+The export engine builds a fresh set of providers for each export job. Each provider is responsible for one CSV file.
 
 ### 1. Implement the `DataProvider` interface
 
@@ -150,13 +150,20 @@ func (p *MyProvider) Collect(ctx context.Context, userID string) ([][]string, er
 }
 ```
 
-### 3. Register in `cmd/main.go`
+### 3. Add it to the per-job provider factory in `cmd/main.go`
+
+Export providers are built fresh for each job by a `ProviderFactory` (a `func(finance financepb.FinanceServiceClient) []engine.DataProvider`) passed to `engine.NewEngine`. Add your provider to the slice it returns; the slice order is the ZIP order:
 
 ```go
-registry.Register(providers.NewMyProvider(myClient))
+newExportProviders := func(finance financepb.FinanceServiceClient) []engine.DataProvider {
+    return []engine.DataProvider{
+        // ...existing providers...
+        providers.NewMyProvider(myClient),
+    }
+}
 ```
 
-The engine handles everything else: CSV writing, ZIP assembly, email delivery, error handling, and metrics emission per provider.
+A finance-backed provider takes the `finance` parameter: a per-job `MemoizedFinanceClient` that collapses `GetAllUserData` to a single call per export and is never shared across jobs. The engine handles everything else: concurrent collection (`errgroup` fan-out), CSV writing, ZIP assembly, email delivery, error handling, and metrics emission per provider.
 
 ### Formatting helpers
 

services/datarights/Dockerfile

@@ -15,6 +15,7 @@ COPY expense/go.mod expense/go.sum* ./expense/
 COPY finance/go.mod finance/go.sum* ./finance/
 COPY healthcheck/go.mod ./healthcheck/
 COPY metrics/go.mod metrics/go.sum* ./metrics/
+COPY perf/go.mod ./perf/
 RUN --mount=type=cache,target=/go/pkg/mod \
     cd datarights && GOWORK=off go mod download
 
@@ -27,6 +28,7 @@ COPY expense/ ./expense/
 COPY finance/ ./finance/
 COPY healthcheck/ ./healthcheck/
 COPY metrics/ ./metrics/
+COPY perf/ ./perf/
 
 RUN --mount=type=cache,target=/go/pkg/mod \
     --mount=type=cache,target=/root/.cache/go-build \

services/datarights/cmd/main.go

@@ -134,21 +134,29 @@ func run() error {
 	// Build dependency graph
 	repo := repository.NewPostgresJobRepository(pool)
 
-	// Set up export engine with provider registry
-	registry := engine.NewProviderRegistry()
-	registry.Register(providers.NewProfileProvider(authClient))
-	registry.Register(providers.NewExpensesProvider(expenseClient, financeClient))
-	registry.Register(providers.NewTagsProvider(financeClient))
-	registry.Register(providers.NewBudgetPeriodsProvider(financeClient))
-	registry.Register(providers.NewDefaultSettingsProvider(financeClient))
+	// Set up export engine with a per-job provider factory. The factory closes
+	// over the auth and expense clients; the finance-backed providers receive a
+	// fresh per-job MemoizedFinanceClient (built inside the engine) so a single
+	// GetAllUserData call is shared across providers without leaking data across
+	// jobs. Registration/ZIP order: profile, expenses, tags, budget_periods,
+	// default_settings.
+	newExportProviders := func(finance financepb.FinanceServiceClient) []engine.DataProvider {
+		return []engine.DataProvider{
+			providers.NewProfileProvider(authClient),
+			providers.NewExpensesProvider(expenseClient, finance),
+			providers.NewTagsProvider(finance),
+			providers.NewBudgetPeriodsProvider(finance),
+			providers.NewDefaultSettingsProvider(finance),
+		}
+	}
 
 	// Set up email sender
 	emailSender, err := buildEmailSender(cfg, logger)
 	if err != nil {
 		return fmt.Errorf("setting up email sender: %w", err)
 	}
 
-	exportEngine := engine.NewEngine(registry, repo, emailSender, cfg.MaxConcurrent, cfg.ExportTimeout, logger)
+	exportEngine := engine.NewEngine(newExportProviders, financeClient, repo, emailSender, cfg.MaxConcurrent, cfg.ExportTimeout, logger)
 
 	// Startup recovery: re-submit non-terminal export jobs
 	emailResolver := service.NewAuthUserEmailResolver(authClient)

services/datarights/go.mod

@@ -10,10 +10,12 @@ require (
 	github.com/ItsThompson/gofin/services/finance v0.0.0
 	github.com/ItsThompson/gofin/services/healthcheck v0.0.0
 	github.com/ItsThompson/gofin/services/metrics v0.0.0
+	github.com/ItsThompson/gofin/services/perf v0.0.0
 	github.com/gin-gonic/gin v1.12.0
 	github.com/jackc/pgx/v5 v5.9.2
 	github.com/prometheus/client_golang v1.22.0
 	github.com/stretchr/testify v1.11.1
+	golang.org/x/sync v0.20.0
 	google.golang.org/grpc v1.80.0
 )
 
@@ -31,6 +33,8 @@ replace github.com/ItsThompson/gofin/services/healthcheck => ../healthcheck
 
 replace github.com/ItsThompson/gofin/services/metrics => ../metrics
 
+replace github.com/ItsThompson/gofin/services/perf => ../perf
+
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bytedance/gopkg v0.1.3 // indirect
@@ -76,7 +80,6 @@ require (
 	golang.org/x/arch v0.22.0 // indirect
 	golang.org/x/crypto v0.50.0 // indirect
 	golang.org/x/net v0.52.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.43.0 // indirect
 	golang.org/x/text v0.36.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260120221211-b8f7ae30c516 // indirect

services/datarights/internal/deletion/providers/mock_clients_test.go

@@ -116,9 +116,6 @@ func (m *mockExpenseClient) GetExpense(_ context.Context, _ *expensepb.GetExpens
 func (m *mockExpenseClient) CountExpensesByTag(_ context.Context, _ *expensepb.CountExpensesByTagRequest, _ ...grpc.CallOption) (*expensepb.CountExpensesByTagResponse, error) {
 	return nil, nil
 }
-func (m *mockExpenseClient) GetAllUserExpenses(_ context.Context, _ *expensepb.GetAllUserExpensesRequest, _ ...grpc.CallOption) (*expensepb.ExpenseListResponse, error) {
-	return nil, nil
-}
 func (m *mockExpenseClient) CorrectExpense(_ context.Context, _ *expensepb.CorrectExpenseRequest, _ ...grpc.CallOption) (*expensepb.ExpenseResponse, error) {
 	return nil, nil
 }
@@ -128,6 +125,9 @@ func (m *mockExpenseClient) GetCorrectionHistory(_ context.Context, _ *expensepb
 func (m *mockExpenseClient) GetProRataGroup(_ context.Context, _ *expensepb.GetProRataGroupRequest, _ ...grpc.CallOption) (*expensepb.ExpenseListResponse, error) {
 	return nil, nil
 }
+func (m *mockExpenseClient) StreamAllUserExpenses(_ context.Context, _ *expensepb.StreamAllUserExpensesRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[expensepb.ExpenseData], error) {
+	return nil, nil
+}
 
 // ---------------------------------------------------------------------------
 // Mock auth client

services/datarights/internal/engine/engine.go

@@ -2,42 +2,57 @@ package engine
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"strings"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/ItsThompson/gofin/services/datarights/internal/email"
 	exportmetrics "github.com/ItsThompson/gofin/services/datarights/internal/metrics"
 	"github.com/ItsThompson/gofin/services/datarights/internal/repository"
+	"github.com/ItsThompson/gofin/services/finance/proto/financepb"
 )
 
+// ProviderFactory builds a fresh set of data providers for a single export job,
+// injecting the finance client the finance-backed providers should share. The
+// factory closes over the non-finance clients (auth, expense) at startup; the
+// finance client is supplied per job so each job gets its own memoized instance.
+type ProviderFactory func(finance financepb.FinanceServiceClient) []DataProvider
+
 // Engine manages a bounded pool of export workers.
 type Engine struct {
-	registry *ProviderRegistry
-	repo     repository.JobRepository
-	sender   email.Sender
-	sem      chan struct{}
-	logger   *slog.Logger
-	timeout  time.Duration
+	newProviders  ProviderFactory
+	financeClient financepb.FinanceServiceClient
+	repo          repository.JobRepository
+	sender        email.Sender
+	sem           chan struct{}
+	logger        *slog.Logger
+	timeout       time.Duration
 }
 
-// NewEngine creates an export engine with bounded concurrency.
+// NewEngine creates an export engine with bounded concurrency. newProviders
+// builds a fresh provider set per job; financeClient is the raw finance client
+// wrapped in a per-job MemoizedFinanceClient before being handed to the factory.
 func NewEngine(
-	registry *ProviderRegistry,
+	newProviders ProviderFactory,
+	financeClient financepb.FinanceServiceClient,
 	repo repository.JobRepository,
 	sender email.Sender,
 	maxConcurrent int,
 	timeout time.Duration,
 	logger *slog.Logger,
 ) *Engine {
 	return &Engine{
-		registry: registry,
-		repo:     repo,
-		sender:   sender,
-		sem:      make(chan struct{}, maxConcurrent),
-		logger:   logger,
-		timeout:  timeout,
+		newProviders:  newProviders,
+		financeClient: financeClient,
+		repo:          repo,
+		sender:        sender,
+		sem:           make(chan struct{}, maxConcurrent),
+		logger:        logger,
+		timeout:       timeout,
 	}
 }
 
@@ -99,53 +114,80 @@ func (e *Engine) execute(ctx context.Context, jobID, userID, userEmail string) {
 		return
 	}
 
-	// Collect data from all providers
-	var csvFiles []CSVFile
-	for _, provider := range e.registry.All() {
-		// Check context before each provider
-		if err := ctx.Err(); err != nil {
-			e.failJob(ctx, jobID, userID, "Export timed out", "collection", jobStart)
-			return
-		}
-
-		e.logger.Debug("provider collection started",
-			slog.String("job_id", jobID),
-			slog.String("user_id", userID),
-			slog.String("provider", provider.Name()),
-			slog.String("method", "engine.execute"),
-		)
-
-		providerStart := time.Now()
-		rows, err := provider.Collect(ctx, userID)
-		providerDuration := time.Since(providerStart).Seconds()
-
-		exportmetrics.ExportDataCollectionDurationSeconds.WithLabelValues(provider.Name()).Observe(providerDuration)
-
-		if err != nil {
-			if ctx.Err() != nil {
-				e.failJob(ctx, jobID, userID, "Export timed out", "collection", jobStart)
-				return
+	// Build a fresh provider set for this job. The finance-backed providers
+	// share one per-job MemoizedFinanceClient, so GetAllUserData is fetched at
+	// most once; a fresh instance per job prevents cross-user data leakage.
+	fc := NewMemoizedFinanceClient(e.financeClient)
+	providerSet := e.newProviders(fc)
+
+	// Collect from every provider concurrently. The providers are independent
+	// and read-only, so each goroutine writes only its own pre-assigned index in
+	// csvFiles; this makes collection latency max(providers) instead of sum while
+	// keeping ZIP order deterministic (the factory's slice order).
+	csvFiles := make([]CSVFile, len(providerSet))
+	g, gctx := errgroup.WithContext(ctx)
+	for i, provider := range providerSet {
+		i, provider := i, provider
+		g.Go(func() error {
+			e.logger.Debug("provider collection started",
+				slog.String("job_id", jobID),
+				slog.String("user_id", userID),
+				slog.String("provider", provider.Name()),
+				slog.String("method", "engine.execute"),
+			)
+
+			providerStart := time.Now()
+			rows, err := provider.Collect(gctx, userID)
+			providerDuration := time.Since(providerStart).Seconds()
+
+			// HistogramVec is goroutine-safe, so the per-provider observation
+			// stays inside the goroutine.
+			exportmetrics.ExportDataCollectionDurationSeconds.WithLabelValues(provider.Name()).Observe(providerDuration)
+
+			if err != nil {
+				// Attach the provider name before errgroup captures the error:
+				// Wait surfaces only the first error, and the post-Wait switch
+				// recovers the name via errors.As on *collectError.
+				return &collectError{provider: provider.Name(), err: err}
 			}
-			e.failJob(ctx, jobID, userID, fmt.Sprintf("Failed to collect %s data", provider.Name()), "collection", jobStart)
-			return
-		}
 
-		e.logger.Info("provider collection complete",
-			slog.String("job_id", jobID),
-			slog.String("user_id", userID),
-			slog.String("provider", provider.Name()),
-			slog.Int("row_count", len(rows)),
-			slog.Float64("duration_seconds", providerDuration),
-			slog.String("method", "engine.execute"),
-		)
-
-		csvFiles = append(csvFiles, CSVFile{
-			Name:    provider.Name() + ".csv",
-			Headers: provider.Headers(),
-			Rows:    rows,
+			e.logger.Info("provider collection complete",
+				slog.String("job_id", jobID),
+				slog.String("user_id", userID),
+				slog.String("provider", provider.Name()),
+				slog.Int("row_count", len(rows)),
+				slog.Float64("duration_seconds", providerDuration),
+				slog.String("method", "engine.execute"),
+			)
+
+			// Write only this goroutine's own slot; never append (data race).
+			csvFiles[i] = CSVFile{
+				Name:    provider.Name() + ".csv",
+				Headers: provider.Headers(),
+				Rows:    rows,
+			}
+			return nil
 		})
 	}
 
+	// Fan-in barrier: the first error wins and cancels the siblings via gctx.
+	// Recheck the job context afterward so a deadline still maps to "Export timed
+	// out" while a genuine provider failure maps to "Failed to collect X data".
+	var ce *collectError
+	switch err := g.Wait(); {
+	case err == nil:
+		// all providers succeeded; fall through to ZIP assembly
+	case ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded):
+		e.failJob(ctx, jobID, userID, "Export timed out", "collection", jobStart)
+		return
+	case errors.As(err, &ce):
+		e.failJob(ctx, jobID, userID, fmt.Sprintf("Failed to collect %s data", ce.provider), "collection", jobStart)
+		return
+	default:
+		e.failJob(ctx, jobID, userID, "Failed to collect export data", "collection", jobStart)
+		return
+	}
+
 	// Build ZIP
 	zipBytes, err := BuildZIP(csvFiles)
 	if err != nil {
@@ -229,6 +271,19 @@ func (e *Engine) failJob(_ context.Context, jobID, userID, errMsg, stage string,
 	}
 }
 
+// collectError carries the failing provider's name alongside the underlying
+// cause, so the post-Wait switch recovers the name via errors.As instead of
+// round-tripping structured data through the error string. It unwraps to the
+// cause so errors.Is(err, context.DeadlineExceeded) still classifies timeouts.
+type collectError struct {
+	provider string
+	err      error
+}
+
+func (e *collectError) Error() string { return "collect " + e.provider + ": " + e.err.Error() }
+
+func (e *collectError) Unwrap() error { return e.err }
+
 // sanitizeError extracts a human-readable reason from an error without exposing internals.
 func sanitizeError(err error) string {
 	msg := err.Error()