fix(purchase): pin each execution lineage's fan-out shape

internal/analytics/collector_test.go

@@ -839,6 +839,12 @@ func (m *mockConfigStore) SaveExecutionIfStatus(ctx context.Context, e *config.P
 func (m *mockConfigStore) SaveExecutionIfStatusTx(ctx context.Context, _ pgx.Tx, e *config.PurchaseExecution, _ []string) error {
 	return m.SavePurchaseExecution(ctx, e)
 }
+func (m *mockConfigStore) GetExecutionFanOutShape(_ context.Context, _ string) (string, error) {
+	return "", nil
+}
+func (m *mockConfigStore) PinExecutionFanOutShape(_ context.Context, _ string, shape string) (string, error) {
+	return shape, nil
+}
 func (m *mockConfigStore) RecoverReapedExecution(ctx context.Context, e *config.PurchaseExecution) error {
 	return m.SavePurchaseExecution(ctx, e)
 }

internal/config/interfaces.go

@@ -138,6 +138,13 @@ type StoreInterface interface {
 	// outcome over a reaper-failed row that has no retry successor linked
 	// (ErrExecutionRetryLinked when one is).
 	RecoverReapedExecution(ctx context.Context, execution *PurchaseExecution) error
+	// PinExecutionFanOutShape records shape (FanOutShapeSingle or
+	// FanOutShapeMulti) for the execution lineage lineageKey unless one is
+	// already recorded, and returns the recorded shape (issue #97).
+	PinExecutionFanOutShape(ctx context.Context, lineageKey, shape string) (string, error)
+	// GetExecutionFanOutShape returns the shape pinned for lineageKey, or ""
+	// when none is pinned yet (issue #97).
+	GetExecutionFanOutShape(ctx context.Context, lineageKey string) (string, error)
 	GetPendingExecutions(ctx context.Context) ([]PurchaseExecution, error)
 	// GetExecutionsByStatuses returns executions in any of the given states,
 	// newest first, capped at `limit`. Used by the History handler to render

internal/config/store_postgres_fan_out_shape.go

@@ -0,0 +1,60 @@
+package config
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// Fan-out shapes of a purchase execution lineage (issue #97). The shape decides
+// which key the per-rec provider idempotency tokens derive from: the lineage key
+// itself on the single-account path, "<lineage key>:<account id>" per account on
+// the fan-out path.
+const (
+	FanOutShapeSingle = "single"
+	FanOutShapeMulti  = "multi"
+)
+
+// PinExecutionFanOutShape records shape as the fan-out shape of the execution
+// lineage keyed by lineageKey unless one is already recorded, and returns the
+// recorded shape. The first writer wins and the row is never updated, so every
+// later attempt of the lineage gets back the shape its first attempt ran with,
+// however the plan's accounts have changed since.
+func (s *PostgresStore) PinExecutionFanOutShape(ctx context.Context, lineageKey, shape string) (string, error) {
+	if lineageKey == "" {
+		return "", fmt.Errorf("cannot pin a fan-out shape for an empty lineage key")
+	}
+	if shape != FanOutShapeSingle && shape != FanOutShapeMulti {
+		return "", fmt.Errorf("cannot pin unknown fan-out shape %q for lineage %s", shape, lineageKey)
+	}
+	// DO UPDATE with a no-op SET (rather than DO NOTHING) so RETURNING yields
+	// the stored row on conflict, in one statement that also waits out a
+	// concurrent first insert of the same key.
+	var pinned string
+	err := s.db.QueryRow(ctx, `
+		INSERT INTO purchase_execution_fan_out_shapes (idempotency_key, shape)
+		VALUES ($1, $2)
+		ON CONFLICT (idempotency_key) DO UPDATE SET shape = purchase_execution_fan_out_shapes.shape
+		RETURNING shape`, lineageKey, shape).Scan(&pinned)
+	if err != nil {
+		return "", fmt.Errorf("failed to pin fan-out shape for lineage %s: %w", lineageKey, err)
+	}
+	return pinned, nil
+}
+
+// GetExecutionFanOutShape returns the fan-out shape pinned for the execution
+// lineage keyed by lineageKey, or "" when none is pinned yet.
+func (s *PostgresStore) GetExecutionFanOutShape(ctx context.Context, lineageKey string) (string, error) {
+	var shape string
+	err := s.db.QueryRow(ctx,
+		`SELECT shape FROM purchase_execution_fan_out_shapes WHERE idempotency_key = $1`, lineageKey).Scan(&shape)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("failed to read fan-out shape for lineage %s: %w", lineageKey, err)
+	}
+	return shape, nil
+}

internal/config/store_postgres_fan_out_shape_integration_test.go

@@ -0,0 +1,68 @@
+//go:build integration
+// +build integration
+
+package config
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestPinExecutionFanOutShape pins the issue #97 store contract against a real
+// PostgreSQL: the first shape pinned for a lineage is returned to every later
+// caller, including concurrent first callers, and invalid input is rejected.
+func TestPinExecutionFanOutShape(t *testing.T) {
+	ctx := context.Background()
+	store := setupRampStepStore(ctx, t)
+
+	t.Run("first writer wins", func(t *testing.T) {
+		key := uuid.NewString()
+		unpinned, err := store.GetExecutionFanOutShape(ctx, key)
+		require.NoError(t, err)
+		assert.Empty(t, unpinned, "a lineage nothing pinned reads as empty")
+
+		got, err := store.PinExecutionFanOutShape(ctx, key, FanOutShapeSingle)
+		require.NoError(t, err)
+		assert.Equal(t, FanOutShapeSingle, got)
+
+		got, err = store.PinExecutionFanOutShape(ctx, key, FanOutShapeMulti)
+		require.NoError(t, err)
+		assert.Equal(t, FanOutShapeSingle, got, "a later attempt must get the first attempt's shape back")
+
+		read, err := store.GetExecutionFanOutShape(ctx, key)
+		require.NoError(t, err)
+		assert.Equal(t, FanOutShapeSingle, read, "the read must return the pinned shape")
+	})
+
+	t.Run("concurrent first attempts agree", func(t *testing.T) {
+		key := uuid.NewString()
+		shapes := []string{FanOutShapeSingle, FanOutShapeMulti, FanOutShapeSingle, FanOutShapeMulti}
+		got := make([]string, len(shapes))
+		var wg sync.WaitGroup
+		for i := range shapes {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				s, err := store.PinExecutionFanOutShape(ctx, key, shapes[i])
+				assert.NoError(t, err)
+				got[i] = s
+			}(i)
+		}
+		wg.Wait()
+		for i := range got {
+			assert.Equal(t, got[0], got[i], "every concurrent caller must see the one pinned shape")
+		}
+	})
+
+	t.Run("invalid input", func(t *testing.T) {
+		_, err := store.PinExecutionFanOutShape(ctx, "", FanOutShapeSingle)
+		assert.Error(t, err)
+		_, err = store.PinExecutionFanOutShape(ctx, uuid.NewString(), "both")
+		assert.Error(t, err)
+	})
+}

internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql

@@ -0,0 +1,7 @@
+-- Reverse 000123: drop the fan-out shape pins (issue #97).
+--
+-- Lossy by design: every pinned shape is dropped, and re-applying the up
+-- migration restores only the "multi" pins its backfill can prove from
+-- per-account rows. "single" pins are not recoverable, so their lineages fall
+-- back to the live plan accounts until their next submission pins them again.
+DROP TABLE IF EXISTS purchase_execution_fan_out_shapes;

internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.up.sql

@@ -0,0 +1,32 @@
+-- Migration 000123 (issue #97): pin the fan-out shape of each purchase
+-- execution lineage.
+--
+-- The per-rec provider idempotency token is derived from the lineage key on the
+-- single-account path and from "<lineage key>:<account id>" on the per-account
+-- fan-out path. The executor used to choose the path from the plan's CURRENT
+-- plan_accounts, so a plan that gained its first account (or lost its last one)
+-- between an attempt and its Retry derived a token the provider had never seen
+-- and bought the commitment again.
+--
+-- One row per lineage key (purchase_executions.idempotency_key, or the
+-- execution_id for rows that predate migration 000066), written once by the
+-- first execution attempt and never updated: every later attempt of the same
+-- lineage takes the recorded shape instead of the live plan configuration.
+CREATE TABLE IF NOT EXISTS purchase_execution_fan_out_shapes (
+    idempotency_key TEXT PRIMARY KEY,
+    shape TEXT NOT NULL CHECK (shape IN ('single', 'multi')),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Backfill only what the data proves. A per-account row's lineage key is
+-- "<root key>:<account id>" and only the fan-out path writes one, so its root
+-- lineage fanned out. A root lineage with no per-account row is left unpinned
+-- (its next attempt pins the live shape, the pre-#97 behavior): a fan-out
+-- reaped before any per-account row was saved leaves no such row, so inferring
+-- "single" from their absence could pin the wrong shape.
+INSERT INTO purchase_execution_fan_out_shapes (idempotency_key, shape)
+SELECT DISTINCT split_part(idempotency_key, ':', 1), 'multi'
+  FROM purchase_executions
+ WHERE idempotency_key LIKE '%:%'
+   AND split_part(idempotency_key, ':', 1) <> ''
+ON CONFLICT (idempotency_key) DO NOTHING;

internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes_test.go

@@ -0,0 +1,68 @@
+//go:build integration
+// +build integration
+
+package migrations_test
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestMigration_PurchaseExecutionFanOutShapes pins migration 000123 (issue #97)
+// against a real PostgreSQL: a lineage with a per-account row is backfilled as
+// "multi", a lineage without one is left unpinned, the up statement is
+// idempotent, and the down migration drops the table.
+func TestMigration_PurchaseExecutionFanOutShapes(t *testing.T) {
+	ctx := context.Background()
+	migrationsPath := getMigrationsPath()
+
+	container, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	defer container.Cleanup(ctx)
+	pool := container.DB.Pool()
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 123)))
+
+	fannedOut, singleOnly := uuid.NewString(), uuid.NewString()
+	for _, key := range []string{fannedOut, fannedOut + ":" + uuid.NewString(), singleOnly} {
+		_, err = pool.Exec(ctx, `
+			INSERT INTO purchase_executions (status, scheduled_date, idempotency_key)
+			VALUES ('failed', NOW(), $1)`, key)
+		require.NoError(t, err)
+	}
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 123))
+
+	var shape string
+	require.NoError(t, pool.QueryRow(ctx,
+		`SELECT shape FROM purchase_execution_fan_out_shapes WHERE idempotency_key = $1`, fannedOut).Scan(&shape))
+	assert.Equal(t, "multi", shape, "a lineage with a per-account row fanned out")
+
+	var n int
+	require.NoError(t, pool.QueryRow(ctx,
+		`SELECT count(*) FROM purchase_execution_fan_out_shapes WHERE idempotency_key = $1`, singleOnly).Scan(&n))
+	assert.Zero(t, n, "a lineage without a per-account row must not be guessed")
+	require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM purchase_execution_fan_out_shapes`).Scan(&n))
+	assert.Equal(t, 1, n, "only the root lineage is pinned, never a per-account key")
+
+	up, err := os.ReadFile(filepath.Join(migrationsPath, "000123_purchase_execution_fan_out_shapes.up.sql"))
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, string(up))
+	require.NoError(t, err, "the up statement must be idempotent")
+
+	_, err = pool.Exec(ctx, `INSERT INTO purchase_execution_fan_out_shapes (idempotency_key, shape) VALUES ($1, 'both')`, uuid.NewString())
+	assert.Error(t, err, "only single and multi are valid shapes")
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 123)))
+	var exists bool
+	require.NoError(t, pool.QueryRow(ctx, `SELECT to_regclass('purchase_execution_fan_out_shapes') IS NOT NULL`).Scan(&exists))
+	assert.False(t, exists, "down migration must drop the table")
+}

internal/mocks/fan_out_shape.go

@@ -0,0 +1,64 @@
+package mocks
+
+import (
+	"context"
+	"sync"
+)
+
+// fanOutShapePins is the in-memory stand-in for the
+// purchase_execution_fan_out_shapes table: the first shape pinned for a lineage
+// key is kept and returned to every later caller. The zero value is ready.
+type fanOutShapePins struct {
+	mu   sync.Mutex
+	pins map[string]string
+}
+
+func (p *fanOutShapePins) pin(lineageKey, shape string) string {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if pinned, ok := p.pins[lineageKey]; ok {
+		return pinned
+	}
+	if p.pins == nil {
+		p.pins = make(map[string]string)
+	}
+	p.pins[lineageKey] = shape
+	return shape
+}
+
+func (p *fanOutShapePins) get(lineageKey string) string {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.pins[lineageKey]
+}
+
+// PinExecutionFanOutShape mocks the fan-out shape pin (issue #97).
+// PinExecutionFanOutShapeFn takes priority; with no expectation registered it
+// behaves like the real store (first writer wins per lineage key), so one mock
+// shared across successive attempts of a lineage sees what the first pinned.
+func (m *MockConfigStore) PinExecutionFanOutShape(ctx context.Context, lineageKey, shape string) (string, error) {
+	m.record("PinExecutionFanOutShape", ctx, lineageKey, shape)
+	if m.PinExecutionFanOutShapeFn != nil {
+		return m.PinExecutionFanOutShapeFn(ctx, lineageKey, shape)
+	}
+	if !isExpected(&m.Mock, "PinExecutionFanOutShape") {
+		return m.fanOutShapes.pin(lineageKey, shape), nil
+	}
+	args := m.Called(ctx, lineageKey, shape)
+	return args.String(0), args.Error(1)
+}
+
+// GetExecutionFanOutShape mocks the pinned-shape read (issue #97).
+// GetExecutionFanOutShapeFn takes priority; with no expectation registered it
+// reads what PinExecutionFanOutShape's default pinned on this mock.
+func (m *MockConfigStore) GetExecutionFanOutShape(ctx context.Context, lineageKey string) (string, error) {
+	m.record("GetExecutionFanOutShape", ctx, lineageKey)
+	if m.GetExecutionFanOutShapeFn != nil {
+		return m.GetExecutionFanOutShapeFn(ctx, lineageKey)
+	}
+	if !isExpected(&m.Mock, "GetExecutionFanOutShape") {
+		return m.fanOutShapes.get(lineageKey), nil
+	}
+	args := m.Called(ctx, lineageKey)
+	return args.String(0), args.Error(1)
+}

internal/mocks/stores.go

@@ -52,11 +52,16 @@ type MockConfigStore struct {
 	CountPendingExecutionsForAccountFn  func(ctx context.Context, accountID string) (int, error)
 	ListPendingExecutionIDsForAccountFn func(ctx context.Context, accountID string) ([]string, error)
 	SavePurchaseExecutionFn             func(ctx context.Context, exec *config.PurchaseExecution) error
+	PinExecutionFanOutShapeFn           func(ctx context.Context, lineageKey, shape string) (string, error)
+	GetExecutionFanOutShapeFn           func(ctx context.Context, lineageKey string) (string, error)
 	GetUserEmailByIDFn                  func(ctx context.Context, userID string) (string, error)
 	mock.Mock
 	// callLog records every invocation, including the ones served from a
 	// default without reaching mock.Called. See assertions.go.
 	callLog callLog
+	// fanOutShapes backs PinExecutionFanOutShape's default: first writer wins
+	// per lineage key, like the real table.
+	fanOutShapes fanOutShapePins
 }
 
 // GetGlobalConfig mocks the GetGlobalConfig operation. Returns an empty