diff --git a/internal/testutil/postgres.go b/internal/testutil/postgres.go index 8f7e1105..a27626e4 100644 --- a/internal/testutil/postgres.go +++ b/internal/testutil/postgres.go @@ -1,6 +1,7 @@ package testutil import ( + "runtime" "testing" "time" @@ -52,3 +53,17 @@ func CreatePostgres(t *testing.T) *pgxpool.Pool { require.NoError(t, migrations.Up(ctx, pool, zap.NewNop())) return pool } + +// PostgresOrSkip is [CreatePostgres] for a test that runs only where Docker +// is: without it the test skips, except on a Linux CI runner, where Docker is +// expected and its absence fails the test. +func PostgresOrSkip(t *testing.T) *pgxpool.Pool { + t.Helper() + if IsRunningInCI(t) && runtime.GOOS == "linux" && !IsDockerAvailable(t) { + t.Fatalf("docker is expected in CI linux testing environments, but wasn't found") + } + if !IsDockerAvailable(t) { + t.SkipNow() + } + return CreatePostgres(t) +} diff --git a/internal/testutil/wait.go b/internal/testutil/wait.go new file mode 100644 index 00000000..b6805e5e --- /dev/null +++ b/internal/testutil/wait.go @@ -0,0 +1,48 @@ +package testutil + +import ( + "testing" + "time" +) + +// MinBlock is how long a call that must wait on a writer has to stay blocked. +const MinBlock = 300 * time.Millisecond + +// WaitTimeout is how long a call that must finish is given before the test fails. +const WaitTimeout = 10 * time.Second + +// RequireWaitsForWriter runs write in the background; write must close +// entered once it holds its lock, then block until release is closed before +// finishing. It then starts wait, requires that wait has not returned within [MinBlock], +// releases the writer, and returns the writer's and wait's errors. +func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, release <-chan struct{}) error, wait func() error) (writeErr, waitErr error) { + t.Helper() + entered := make(chan struct{}) + release := make(chan struct{}) + wrote := make(chan error, 1) + go func() { wrote <- write(entered, release) }() + select { + case <-entered: + case err := <-wrote: + t.Fatalf("the writer returned before taking its lock: %v", err) + case <-time.After(WaitTimeout): + t.Fatal("the writer did not take its lock") + } + + waited := make(chan error, 1) + go func() { waited <- wait() }() + select { + case <-waited: + t.Fatal("the waiting call returned while the writer held its lock") + case <-time.After(MinBlock): + } + + close(release) + writeErr = <-wrote + select { + case waitErr = <-waited: + case <-time.After(WaitTimeout): + t.Fatal("the waiting call did not return after the writer finished") + } + return writeErr, waitErr +} diff --git a/pkg/api/service/tenant/service.go b/pkg/api/service/tenant/service.go index 933efcd2..87ae4035 100644 --- a/pkg/api/service/tenant/service.go +++ b/pkg/api/service/tenant/service.go @@ -15,6 +15,7 @@ import ( "github.com/fil-forge/hilt/pkg/store/accesskey" "github.com/fil-forge/hilt/pkg/store/bucket" "github.com/fil-forge/hilt/pkg/store/delegation" + principalstore "github.com/fil-forge/hilt/pkg/store/principal" "github.com/fil-forge/hilt/pkg/store/provider" tenantstore "github.com/fil-forge/hilt/pkg/store/tenant" wrapkeystore "github.com/fil-forge/hilt/pkg/store/wrapkey" @@ -38,6 +39,7 @@ type Service struct { providers provider.Store buckets bucket.Store accessKeys accesskey.Store + principals principalstore.Store delegations delegation.Store secrets vault.Vault wrapKeys wrapkeystore.Store @@ -52,6 +54,7 @@ func New( providers provider.Store, buckets bucket.Store, accessKeys accesskey.Store, + principals principalstore.Store, delegations delegation.Store, secrets vault.Vault, wrapKeys wrapkeystore.Store, @@ -64,6 +67,7 @@ func New( providers: providers, buckets: buckets, accessKeys: accessKeys, + principals: principals, delegations: delegations, secrets: secrets, plcClient: plcClient, @@ -255,8 +259,8 @@ func (s *Service) SetStatus(ctx context.Context, externalID, status string) erro } // Delete permanently deletes a tenant (which must be disabled), cascading to its -// buckets, access keys, and delegations, and deactivating its did:plc. It is -// idempotent: a missing tenant is a no-op. +// buckets, access keys, principals, and delegations, and deactivating its +// did:plc. It is idempotent: a missing tenant is a no-op. // // Out of scope: deprovisioning the tenant's spaces from the Forge upload service // (Sprue), for which there is no facility per the RFC. @@ -299,6 +303,13 @@ func (s *Service) Delete(ctx context.Context, externalID string) error { } } + // Cascade: principals. They reference the tenant row (FK RESTRICT), so they + // go before it. Nothing is published: the tenant is disabled and its keys are + // gone, so no request can reach a principal's cached access. + if err := s.principals.DeleteByTenant(ctx, rec.ID); err != nil { + return fmt.Errorf("deleting principals: %w", err) + } + // Cascade: wrap keys (registry rows + their sealed private halves). Every // version, active and archived, is destroyed — tenant deletion is the // "true deletion" that ends the envelope-recovery path, so archive-don't- diff --git a/pkg/api/service/tenant/service_test.go b/pkg/api/service/tenant/service_test.go index 13821164..5f73d6d1 100644 --- a/pkg/api/service/tenant/service_test.go +++ b/pkg/api/service/tenant/service_test.go @@ -15,6 +15,7 @@ import ( accesskeymemory "github.com/fil-forge/hilt/pkg/store/accesskey/memory" bucketmemory "github.com/fil-forge/hilt/pkg/store/bucket/memory" delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory" + principalmemory "github.com/fil-forge/hilt/pkg/store/principal/memory" providermemory "github.com/fil-forge/hilt/pkg/store/provider/memory" "github.com/fil-forge/hilt/pkg/store/tenant" tenantmemory "github.com/fil-forge/hilt/pkg/store/tenant/memory" @@ -83,7 +84,7 @@ func provisionSetup(t *testing.T, plcStatus int) provisionEnv { require.NoError(t, err) svc := tenantsvc.New(zap.NewNop(), tenantmemory.New(), providers, bucketmemory.New(), - accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), plcClient, upload) + accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), plcClient, upload) return provisionEnv{svc: svc, providers: providers, sprueFailed: sprueFailed} } @@ -144,7 +145,7 @@ func TestProvision(t *testing.T) { // clients — enough for Get and SetStatus, which never touch them. func simpleService(tenants tenant.Store) *tenantsvc.Service { return tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), - accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), nil, nil) + accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), nil, nil) } func TestGetAndSetStatus(t *testing.T) { @@ -215,9 +216,11 @@ func (d *plcDirectory) ServeHTTP(w http.ResponseWriter, r *http.Request) { } type deleteEnv struct { - svc *tenantsvc.Service - tenants *tenantmemory.Store - directory *plcDirectory + svc *tenantsvc.Service + tenants *tenantmemory.Store + principals *principalmemory.Store + tenantID did.DID + directory *plcDirectory } func deleteSetup(t *testing.T, status tenant.Status) deleteEnv { @@ -245,12 +248,14 @@ func deleteSetup(t *testing.T, status tenant.Status) deleteEnv { tenants := tenantmemory.New() secrets := vaultmemory.New() + principals := principalmemory.New() require.NoError(t, tenants.Add(ctx, tenantID, "tenant-1", testutil.RandomDID(t), status)) require.NoError(t, secrets.Write(ctx, vault.TenantKeyPath(tenantID), signer.Bytes())) + require.NoError(t, principals.Add(ctx, tenantID, "user-1")) svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), - accesskeymemory.New(), delegationmemory.New(), secrets, wrapkeysmemory.New(), plcClient, nil) - return deleteEnv{svc: svc, tenants: tenants, directory: directory} + accesskeymemory.New(), principals, delegationmemory.New(), secrets, wrapkeysmemory.New(), plcClient, nil) + return deleteEnv{svc: svc, tenants: tenants, principals: principals, tenantID: tenantID, directory: directory} } func TestDelete(t *testing.T) { @@ -261,6 +266,9 @@ func TestDelete(t *testing.T) { require.NoError(t, env.svc.Delete(ctx, "tenant-1")) _, err := env.tenants.GetByExternalID(ctx, "tenant-1") require.ErrorIs(t, err, store.ErrRecordNotFound) + ps, err := env.principals.ListByTenant(ctx, env.tenantID) + require.NoError(t, err) + require.Empty(t, ps) require.Equal(t, 1, env.directory.deactivations) }) diff --git a/pkg/api/tenants_test.go b/pkg/api/tenants_test.go index da356f24..7182ea30 100644 --- a/pkg/api/tenants_test.go +++ b/pkg/api/tenants_test.go @@ -20,6 +20,7 @@ import ( accesskeymemory "github.com/fil-forge/hilt/pkg/store/accesskey/memory" bucketmemory "github.com/fil-forge/hilt/pkg/store/bucket/memory" delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory" + principalmemory "github.com/fil-forge/hilt/pkg/store/principal/memory" "github.com/fil-forge/hilt/pkg/store/provider" providermemory "github.com/fil-forge/hilt/pkg/store/provider/memory" "github.com/fil-forge/hilt/pkg/store/tenant" @@ -227,7 +228,7 @@ func setupProvision(t *testing.T, cfg *setupConfig) (*echo.Echo, *provisionDeps) upload.WithHTTPClient(&http.Client{Transport: srv})) require.NoError(t, err) - svc := tenantsvc.New(zap.NewNop(), deps.tenants, deps.providers, bucketmemory.New(), accesskeymemory.New(), delegationmemory.New(), deps.secrets, deps.wrapKeys, plcClient, upload) + svc := tenantsvc.New(zap.NewNop(), deps.tenants, deps.providers, bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), deps.secrets, deps.wrapKeys, plcClient, upload) route := api.NewProvisionTenantHandler(zap.NewNop(), svc) e := echo.New() e.Add(route.Method, route.Path, route.Handler) @@ -468,7 +469,7 @@ func TestGetTenantHandler(t *testing.T) { ctx := t.Context() tenants := tenantmemory.New() require.NoError(t, tenants.Add(ctx, testutil.RandomDID(t), "tenant-1", testutil.RandomDID(t), tenant.Active)) - svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil) + svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil) e := serve(api.NewGetTenantHandler(zap.NewNop(), svc)) t.Run("found", func(t *testing.T) { @@ -490,7 +491,7 @@ func TestUpdateTenantStatusHandler(t *testing.T) { tenants := tenantmemory.New() id := testutil.RandomDID(t) require.NoError(t, tenants.Add(ctx, id, "tenant-1", testutil.RandomDID(t), tenant.Active)) - svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil) + svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil) e := serve(api.NewUpdateTenantStatusHandler(zap.NewNop(), svc)) statusBody := func(s api.TenantStatus) []byte { @@ -559,6 +560,7 @@ type deleteDeps struct { tenants *tenantmemory.Store buckets *bucketmemory.Store accessKeys *accesskeymemory.Store + principals *principalmemory.Store delegations *delegationmemory.Store wrapKeys *wrapkeymemory.Store secrets vault.Vault @@ -613,6 +615,7 @@ func setupDelete(t *testing.T, status tenant.Status) (*echo.Echo, *deleteDeps) { tenants: tenantmemory.New(), buckets: bucketmemory.New(), accessKeys: accesskeymemory.New(), + principals: principalmemory.New(), delegations: delegationmemory.New(), wrapKeys: wrapkeymemory.New(), secrets: vaultmemory.New(), @@ -636,7 +639,7 @@ func setupDelete(t *testing.T, status tenant.Status) (*echo.Echo, *deleteDeps) { VaultKey: wrapkeystore.VaultKey(tenantID, 1), })) - svc := tenantsvc.New(zap.NewNop(), deps.tenants, providermemory.New(), deps.buckets, deps.accessKeys, deps.delegations, deps.secrets, deps.wrapKeys, plcClient, nil) + svc := tenantsvc.New(zap.NewNop(), deps.tenants, providermemory.New(), deps.buckets, deps.accessKeys, deps.principals, deps.delegations, deps.secrets, deps.wrapKeys, plcClient, nil) route := api.NewDeleteTenantHandler(zap.NewNop(), svc) return serve(route), deps } @@ -663,6 +666,7 @@ func TestDeleteTenantHandler(t *testing.T) { require.NoError(t, deps.accessKeys.Add(ctx, akID, deps.tenantID, "k1", nil, []string{"s3:GetObject"}, nil)) akVaultKey := "/tenant/" + deps.tenantID.String() + "/access-key/" + akID.String() require.NoError(t, deps.secrets.Write(ctx, akVaultKey, []byte("ak-key"))) + require.NoError(t, deps.principals.Add(ctx, deps.tenantID, "user-1")) require.NoError(t, deps.delegations.PutBatch(ctx, []ucan.Delegation{makeDelegation(t, deps.tenantID)})) require.NoError(t, deps.delegations.PutBatch(ctx, []ucan.Delegation{makeDelegation(t, akID)})) @@ -694,6 +698,11 @@ func TestDeleteTenantHandler(t *testing.T) { _, err = deps.secrets.Read(ctx, akVaultKey) require.ErrorIs(t, err, vault.ErrNotFound) + // Principals gone. + ps, err := deps.principals.ListByTenant(ctx, deps.tenantID) + require.NoError(t, err) + require.Empty(t, ps) + // Delegations to both the tenant and the access key gone. tenantDlgs, err := deps.delegations.ListByAudience(ctx, deps.tenantID) require.NoError(t, err) diff --git a/pkg/fx/store/memory/provider.go b/pkg/fx/store/memory/provider.go index 914c2d33..0c278980 100644 --- a/pkg/fx/store/memory/provider.go +++ b/pkg/fx/store/memory/provider.go @@ -9,6 +9,8 @@ import ( membucket "github.com/fil-forge/hilt/pkg/store/bucket/memory" "github.com/fil-forge/hilt/pkg/store/delegation" memdelegation "github.com/fil-forge/hilt/pkg/store/delegation/memory" + "github.com/fil-forge/hilt/pkg/store/principal" + memprincipal "github.com/fil-forge/hilt/pkg/store/principal/memory" "github.com/fil-forge/hilt/pkg/store/provider" memprovider "github.com/fil-forge/hilt/pkg/store/provider/memory" "github.com/fil-forge/hilt/pkg/store/tenant" @@ -24,6 +26,7 @@ var Module = fx.Module("memory-store", fx.Annotate(memaccesskey.New, fx.As(new(accesskey.Store))), fx.Annotate(membucket.New, fx.As(new(bucket.Store))), fx.Annotate(memdelegation.New, fx.As(new(delegation.Store))), + fx.Annotate(memprincipal.New, fx.As(new(principal.Store))), fx.Annotate(memprovider.New, fx.As(new(provider.Store))), fx.Annotate(memtenant.New, fx.As(new(tenant.Store))), fx.Annotate(memwrapkey.New, fx.As(new(wrapkey.Store))), diff --git a/pkg/fx/store/postgres/provider.go b/pkg/fx/store/postgres/provider.go index 80621561..693aa7ad 100644 --- a/pkg/fx/store/postgres/provider.go +++ b/pkg/fx/store/postgres/provider.go @@ -16,6 +16,8 @@ import ( pgbucket "github.com/fil-forge/hilt/pkg/store/bucket/postgres" "github.com/fil-forge/hilt/pkg/store/delegation" pgdelegation "github.com/fil-forge/hilt/pkg/store/delegation/postgres" + "github.com/fil-forge/hilt/pkg/store/principal" + pgprincipal "github.com/fil-forge/hilt/pkg/store/principal/postgres" "github.com/fil-forge/hilt/pkg/store/provider" pgprovider "github.com/fil-forge/hilt/pkg/store/provider/postgres" "github.com/fil-forge/hilt/pkg/store/tenant" @@ -35,6 +37,7 @@ var Module = fx.Module("postgres-store", NewAccessKeyStore, NewBucketStore, NewDelegationStore, + NewPrincipalStore, NewProviderStore, NewTenantStore, NewWrapKeyStore, @@ -120,6 +123,10 @@ func NewDelegationStore(mdb *MigratedPool) delegation.Store { return pgdelegation.New(mdb.Pool) } +func NewPrincipalStore(mdb *MigratedPool) principal.Store { + return pgprincipal.New(mdb.Pool) +} + func NewProviderStore(mdb *MigratedPool) provider.Store { return pgprovider.New(mdb.Pool) } diff --git a/pkg/migrations/sql/00006_principal.sql b/pkg/migrations/sql/00006_principal.sql new file mode 100644 index 00000000..a1054668 --- /dev/null +++ b/pkg/migrations/sql/00006_principal.sql @@ -0,0 +1,20 @@ +-- +goose Up +-- +goose StatementBegin +-- principal is a console user within a tenant, identified by the console's +-- principalId. A principal holds no key material and no delegations: its access +-- is computed from bucket policies at request time. Removal sets deleted_at +-- rather than deleting the row, so a later PUT for the same id revives it and +-- an id reused by the console never inherits anything. +CREATE TABLE principal ( + tenant_id TEXT NOT NULL REFERENCES tenant(id) ON DELETE RESTRICT, -- DID (did:plc) + external_id TEXT NOT NULL, -- console principalId + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deleted_at TIMESTAMPTZ, -- set by removal; cleared by revive + PRIMARY KEY (tenant_id, external_id) +); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +DROP TABLE IF EXISTS principal; +-- +goose StatementEnd diff --git a/pkg/store/principal/memory/store.go b/pkg/store/principal/memory/store.go new file mode 100644 index 00000000..774eb455 --- /dev/null +++ b/pkg/store/principal/memory/store.go @@ -0,0 +1,174 @@ +// Package memory provides an in-memory implementation of principal.Store. +// +// The store holds two locks. mutex guards the map and is held for one read or +// one write at a time, never across a caller's callback. removals serializes a +// removal with the writes that must not interleave with it, and is the lock +// Tombstone holds while fn runs. +// +// The split mirrors Postgres. A removal there holds the principal row FOR +// UPDATE across its callback while ListByTenant still reads the table without +// waiting; the callback rewrites the policies naming the principal, and a +// policy write of its own lists this store's principals. Holding one lock +// across both would wedge the two writes against each other. +package memory + +import ( + "context" + "fmt" + "slices" + "strings" + "sync" + "time" + + "github.com/fil-forge/hilt/pkg/store" + "github.com/fil-forge/hilt/pkg/store/principal" + "github.com/fil-forge/ucantone/did" +) + +// entry is a stored principal; deleted marks a tombstone that no read returns +// and that Add revives. +type entry struct { + rec principal.Record + deleted bool +} + +type Store struct { + mutex sync.RWMutex + removals sync.Mutex + principals map[did.DID]map[string]entry +} + +var _ principal.Store = (*Store)(nil) + +func New() *Store { + return &Store{principals: map[did.DID]map[string]entry{}} +} + +func (s *Store) Add(ctx context.Context, tenant did.DID, externalID string) error { + if tenant == did.Undef { + return fmt.Errorf("principal tenant is required: %w", store.ErrInvalidArgument) + } + if externalID == "" { + return fmt.Errorf("principal external ID is required: %w", store.ErrInvalidArgument) + } + + // A revive must not interleave with a removal of the same principal: the + // removal would tombstone the row the revive just wrote. + s.removals.Lock() + defer s.removals.Unlock() + + s.mutex.Lock() + defer s.mutex.Unlock() + + byID, ok := s.principals[tenant] + if !ok { + byID = map[string]entry{} + s.principals[tenant] = byID + } + if e, ok := byID[externalID]; ok && !e.deleted { + return store.ErrRecordExists + } + byID[externalID] = entry{rec: principal.Record{ + Tenant: tenant, + ExternalID: externalID, + CreatedAt: time.Now().UTC(), + }} + return nil +} + +// Get with [store.WithShareLock] waits for an in-flight Tombstone of any principal +// to finish, the way the Postgres read waits on the row held FOR UPDATE. An +// unlocked read takes the map alone and may be answered while a removal's +// callback is still running, as the unlocked Postgres read is answered from +// its snapshot. +func (s *Store) Get(ctx context.Context, tenant did.DID, externalID string, opts ...store.ReadOption) (principal.Record, error) { + if store.NewReadConfig(opts...).Share { + s.removals.Lock() + defer s.removals.Unlock() + } + + s.mutex.RLock() + defer s.mutex.RUnlock() + + e, ok := s.principals[tenant][externalID] + if !ok || e.deleted { + return principal.Record{}, store.ErrRecordNotFound + } + return e.rec, nil +} + +func (s *Store) ListByTenant(ctx context.Context, tenant did.DID) ([]principal.Record, error) { + s.mutex.RLock() + defer s.mutex.RUnlock() + + var recs []principal.Record + for _, e := range s.principals[tenant] { + if !e.deleted { + recs = append(recs, e.rec) + } + } + slices.SortFunc(recs, func(a, b principal.Record) int { + return strings.Compare(a.ExternalID, b.ExternalID) + }) + return recs, nil +} + +// Tombstone holds removals for the whole call and takes the map mutex only to +// read the entry and, at the end, to write the tombstone. fn runs with the map +// unlocked, so it may write the policy store whose own writes +// read this one. A share-locked Get waits on removals and so observes either +// the principal or its tombstone, never a state in between; ListByTenant reads +// the map and never waits, as on Postgres. +func (s *Store) Tombstone(ctx context.Context, tenant did.DID, externalID string, fn func(ctx context.Context) error) error { + s.removals.Lock() + defer s.removals.Unlock() + + s.mutex.RLock() + e, ok := s.principals[tenant][externalID] + s.mutex.RUnlock() + if !ok || e.deleted { + return nil // idempotent: nothing to publish and nothing to remove + } + + if fn != nil { + if err := fn(ctx); err != nil { + return fmt.Errorf("before tombstoning principal: %w", err) + } + } + + s.mutex.Lock() + defer s.mutex.Unlock() + + byID, ok := s.principals[tenant] + if !ok { + return nil // DeleteByTenant took the whole tenant meanwhile + } + e, ok = byID[externalID] + if !ok || e.deleted { + return nil + } + e.deleted = true + byID[externalID] = e + return nil +} + +// WithLock runs fn without holding anything, and a removal of one of the named +// principals may commit while it runs. Holding removals across fn deadlocks: +// the policy write that calls WithLock holds the bucket's policy, and a removal +// holds removals while its callback rewrites that same policy, so each waits +// on what the other holds. +func (s *Store) WithLock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error { + return fn(ctx) +} + +func (s *Store) DeleteByTenant(ctx context.Context, tenant did.DID) error { + // A tenant removal waits for an in-flight principal removal, as it would on + // the row Postgres holds FOR UPDATE. + s.removals.Lock() + defer s.removals.Unlock() + s.mutex.Lock() + defer s.mutex.Unlock() + + delete(s.principals, tenant) + return nil +} diff --git a/pkg/store/principal/postgres/store.go b/pkg/store/principal/postgres/store.go new file mode 100644 index 00000000..2d455f4e --- /dev/null +++ b/pkg/store/principal/postgres/store.go @@ -0,0 +1,199 @@ +// Package postgres provides a PostgreSQL-backed implementation of +// principal.Store. +package postgres + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/fil-forge/hilt/pkg/store" + "github.com/fil-forge/hilt/pkg/store/principal" + "github.com/fil-forge/ucantone/did" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type Store struct { + pool *pgxpool.Pool +} + +var _ principal.Store = (*Store)(nil) + +func New(pool *pgxpool.Pool) *Store { + return &Store{pool: pool} +} + +// Initialize is a no-op. Schema is managed by the shared goose migrations. +func (s *Store) Initialize(ctx context.Context) error { return nil } + +func (s *Store) Add(ctx context.Context, tenant did.DID, externalID string) error { + if tenant == did.Undef { + return fmt.Errorf("principal tenant is required: %w", store.ErrInvalidArgument) + } + if externalID == "" { + return fmt.Errorf("principal external ID is required: %w", store.ErrInvalidArgument) + } + // The upsert revives a tombstone under the same id; it touches nothing when + // the row is live, and the zero row count reports the duplicate. The row + // lock it takes waits for an in-flight Tombstone of the same principal. + tag, err := s.pool.Exec(ctx, ` + INSERT INTO principal (tenant_id, external_id) + VALUES ($1, $2) + ON CONFLICT (tenant_id, external_id) DO UPDATE + SET deleted_at = NULL, created_at = NOW() + WHERE principal.deleted_at IS NOT NULL + `, tenant.String(), externalID) + if err != nil { + return fmt.Errorf("adding principal: %w", err) + } + if tag.RowsAffected() == 0 { + return store.ErrRecordExists + } + return nil +} + +// Get reads the row, with FOR SHARE when [store.WithShareLock] is passed so the +// read waits on a Tombstone that holds the row FOR UPDATE. +func (s *Store) Get(ctx context.Context, tenant did.DID, externalID string, opts ...store.ReadOption) (principal.Record, error) { + query := ` + SELECT tenant_id, external_id, created_at + FROM principal + WHERE tenant_id = $1 AND external_id = $2 AND deleted_at IS NULL + ` + if store.NewReadConfig(opts...).Share { + query += ` FOR SHARE` + } + rec, err := scanRecord(s.pool.QueryRow(ctx, query, tenant.String(), externalID)) + if errors.Is(err, pgx.ErrNoRows) { + return principal.Record{}, store.ErrRecordNotFound + } + if err != nil { + return principal.Record{}, fmt.Errorf("getting principal: %w", err) + } + return rec, nil +} + +func (s *Store) ListByTenant(ctx context.Context, tenant did.DID) ([]principal.Record, error) { + rows, err := s.pool.Query(ctx, ` + SELECT tenant_id, external_id, created_at + FROM principal + WHERE tenant_id = $1 AND deleted_at IS NULL + ORDER BY external_id ASC + `, tenant.String()) + if err != nil { + return nil, fmt.Errorf("listing principals by tenant: %w", err) + } + defer rows.Close() + + var recs []principal.Record + for rows.Next() { + rec, err := scanRecord(rows) + if err != nil { + return nil, fmt.Errorf("scanning principal: %w", err) + } + recs = append(recs, rec) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating principals: %w", err) + } + return recs, nil +} + +// Tombstone runs in one transaction: it locks the row FOR UPDATE, runs beforeCommit +// while holding the lock, sets deleted_at and commits. A locked +// read of the row (see [Store.Get]) waits for the commit or the rollback. +func (s *Store) Tombstone(ctx context.Context, tenant did.DID, externalID string, beforeCommit func(ctx context.Context) error) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("beginning transaction: %w", err) + } + defer tx.Rollback(ctx) // no-op once committed; rolls back on any early return + + var deleted bool + err = tx.QueryRow(ctx, ` + SELECT deleted_at IS NOT NULL + FROM principal + WHERE tenant_id = $1 AND external_id = $2 + FOR UPDATE + `, tenant.String(), externalID).Scan(&deleted) + if errors.Is(err, pgx.ErrNoRows) || (err == nil && deleted) { + return nil // idempotent: nothing to publish and nothing to remove + } + if err != nil { + return fmt.Errorf("locking principal: %w", err) + } + + if beforeCommit != nil { + if err := beforeCommit(ctx); err != nil { + return fmt.Errorf("before tombstoning principal: %w", err) + } + } + + if _, err := tx.Exec(ctx, ` + UPDATE principal + SET deleted_at = NOW() + WHERE tenant_id = $1 AND external_id = $2 + `, tenant.String(), externalID); err != nil { + return fmt.Errorf("tombstoning principal: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } + return nil +} + +// WithLock runs in one transaction: it locks the live rows FOR UPDATE in +// external ID order, runs fn while holding them and commits. Sorting keeps two +// WithLock calls over overlapping principals from deadlocking. +func (s *Store) WithLock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error { + if len(externalIDs) == 0 { + return fn(ctx) + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("beginning transaction: %w", err) + } + defer tx.Rollback(ctx) // no-op once committed; rolls back on any early return + + if _, err := tx.Exec(ctx, ` + SELECT 1 + FROM principal + WHERE tenant_id = $1 AND external_id = ANY($2) AND deleted_at IS NULL + ORDER BY external_id ASC + FOR UPDATE + `, tenant.String(), externalIDs); err != nil { + return fmt.Errorf("locking principals: %w", err) + } + if err := fn(ctx); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } + return nil +} + +func (s *Store) DeleteByTenant(ctx context.Context, tenant did.DID) error { + if _, err := s.pool.Exec(ctx, `DELETE FROM principal WHERE tenant_id = $1`, tenant.String()); err != nil { + return fmt.Errorf("deleting principals by tenant: %w", err) + } + return nil +} + +func scanRecord(row pgx.Row) (principal.Record, error) { + var ( + tenantStr string + externalID string + createdAt time.Time + ) + if err := row.Scan(&tenantStr, &externalID, &createdAt); err != nil { + return principal.Record{}, err + } + tenant, err := did.Parse(tenantStr) + if err != nil { + return principal.Record{}, fmt.Errorf("parsing tenant DID: %w", err) + } + return principal.Record{Tenant: tenant, ExternalID: externalID, CreatedAt: createdAt}, nil +} diff --git a/pkg/store/principal/principal.go b/pkg/store/principal/principal.go new file mode 100644 index 00000000..d5e0a2c1 --- /dev/null +++ b/pkg/store/principal/principal.go @@ -0,0 +1,67 @@ +// Package principal defines the store of a tenant's principals: the console +// users (identified by the console's principalId) whose access to the tenant's +// buckets is computed from bucket policies at request time. A principal holds +// no key material and no delegations. +// +// A removed principal stays in the store as a tombstone that no read returns. +// Add revives it, so the console may reuse an id without the new principal +// inheriting anything from the old one. +package principal + +import ( + "context" + "time" + + "github.com/fil-forge/hilt/pkg/store" + "github.com/fil-forge/ucantone/did" +) + +// Record is one principal of a tenant. +type Record struct { + // Tenant is the tenant DID (did:plc) the principal belongs to. + Tenant did.DID + // ExternalID is the console principalId that identifies the principal + // within the tenant. + ExternalID string + // CreatedAt is when the principal was recorded. + CreatedAt time.Time +} + +// Store persists principal records. +type Store interface { + // Add records a principal, or revives a removed one under the same external + // ID with a fresh CreatedAt. It returns [store.ErrInvalidArgument] if the + // tenant is undef or the external ID is empty, and [store.ErrRecordExists] + // if the tenant already has a live principal with that external ID. + Add(ctx context.Context, tenant did.DID, externalID string) error + // Get returns the tenant's principal with the given external ID. It returns + // [store.ErrRecordNotFound] if there is none or it was removed. With + // [store.WithShareLock] the read waits for an in-flight [Store.Tombstone] of + // the same row to commit or roll back. + Get(ctx context.Context, tenant did.DID, externalID string, opts ...store.ReadOption) (Record, error) + // ListByTenant returns every live principal of the tenant, ordered by + // external ID. + ListByTenant(ctx context.Context, tenant did.DID) ([]Record, error) + // Tombstone marks a principal's row removed; the row stays, so that Add + // can revive it. It excludes concurrent removals and revives of the same + // row for the whole call, runs fn (nil allowed) inside the write, before + // it takes effect, then marks the row. An error from fn is returned and + // leaves the principal live. It is idempotent: when no live row exists it + // returns nil without running fn. + // + // fn must not read or write this store: a share-locked read or a write of + // the same row waits on the lock the call itself holds. It may write other + // stores, including ones whose own writes read this one: no lock on this + // store's records is held while it runs. + Tombstone(ctx context.Context, tenant did.DID, externalID string, fn func(ctx context.Context) error) error + // WithLock runs fn while holding the tenant's live principals with the + // given external IDs against concurrent removals, revives and share-locked + // reads, and releases them when fn returns. A policy write uses it so that a key created for one + // of the principals meanwhile is either included in the write's rotation or + // created from the committed policy. IDs with no live row are skipped. fn + // follows the contract of [Store.Tombstone]'s fn. + WithLock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error + // DeleteByTenant deletes the rows of every principal of the tenant, + // tombstones included. It is idempotent. + DeleteByTenant(ctx context.Context, tenant did.DID) error +} diff --git a/pkg/store/principal/principal_test.go b/pkg/store/principal/principal_test.go new file mode 100644 index 00000000..bf8fbc7f --- /dev/null +++ b/pkg/store/principal/principal_test.go @@ -0,0 +1,408 @@ +package principal_test + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + htestutil "github.com/fil-forge/hilt/internal/testutil" + "github.com/fil-forge/hilt/pkg/store" + "github.com/fil-forge/hilt/pkg/store/principal" + principalmemory "github.com/fil-forge/hilt/pkg/store/principal/memory" + principalpostgres "github.com/fil-forge/hilt/pkg/store/principal/postgres" + providerpostgres "github.com/fil-forge/hilt/pkg/store/provider/postgres" + "github.com/fil-forge/hilt/pkg/store/tenant" + tenantpostgres "github.com/fil-forge/hilt/pkg/store/tenant/postgres" + "github.com/fil-forge/libforge/testutil" + "github.com/fil-forge/ucantone/did" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" +) + +type StoreKind string + +const ( + Memory StoreKind = "memory" + Postgres StoreKind = "postgres" +) + +var storeKinds = []StoreKind{Memory, Postgres} + +// seedFunc ensures the parent tenant (and its provider) exist so the +// principal.tenant_id foreign key is satisfied. It is a no-op for the memory +// store, which does not enforce referential integrity. +type seedFunc func(t *testing.T, tenantID did.DID) + +func makeStore(t *testing.T, k StoreKind) (principal.Store, seedFunc) { + switch k { + case Memory: + return principalmemory.New(), func(*testing.T, did.DID) {} + case Postgres: + pool := htestutil.PostgresOrSkip(t) + return principalpostgres.New(pool), seeder(pool) + } + panic("unknown store kind") +} + +func seeder(pool *pgxpool.Pool) seedFunc { + providers := providerpostgres.New(pool) + tenants := tenantpostgres.New(pool) + return func(t *testing.T, tenantID did.DID) { + providerID := testutil.RandomDID(t) + require.NoError(t, providers.Add(t.Context(), providerID, tenantID.String(), nil)) + require.NoError(t, tenants.Add(t.Context(), tenantID, "ext-"+tenantID.String(), providerID, tenant.Active)) + } +} + +func TestPrincipalStore(t *testing.T) { + for _, k := range storeKinds { + t.Run(string(k), func(t *testing.T) { + s, seed := makeStore(t, k) + + t.Run("adds and retrieves a principal", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "user-1")) + + rec, err := s.Get(t.Context(), tenantID, "user-1") + require.NoError(t, err) + require.Equal(t, tenantID, rec.Tenant) + require.Equal(t, "user-1", rec.ExternalID) + require.False(t, rec.CreatedAt.IsZero()) + }) + + t.Run("Get honours the share lock option", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "user-1")) + + rec, err := s.Get(t.Context(), tenantID, "user-1", store.WithShareLock()) + require.NoError(t, err) + require.Equal(t, "user-1", rec.ExternalID) + }) + + t.Run("Get returns ErrRecordNotFound for an unknown principal", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + _, err := s.Get(t.Context(), tenantID, "nobody") + require.ErrorIs(t, err, store.ErrRecordNotFound) + _, err = s.Get(t.Context(), tenantID, "nobody", store.WithShareLock()) + require.ErrorIs(t, err, store.ErrRecordNotFound) + }) + + t.Run("Get isolates by tenant", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + other := testutil.RandomDID(t) + seed(t, tenantID) + seed(t, other) + require.NoError(t, s.Add(t.Context(), tenantID, "shared-id")) + + _, err := s.Get(t.Context(), other, "shared-id") + require.ErrorIs(t, err, store.ErrRecordNotFound) + }) + + t.Run("Add returns ErrRecordExists for a duplicate (tenant, external ID)", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + other := testutil.RandomDID(t) + seed(t, tenantID) + seed(t, other) + require.NoError(t, s.Add(t.Context(), tenantID, "dup")) + require.ErrorIs(t, s.Add(t.Context(), tenantID, "dup"), store.ErrRecordExists) + // The same external ID under a different tenant is a different principal. + require.NoError(t, s.Add(t.Context(), other, "dup")) + }) + + t.Run("Add returns ErrInvalidArgument for an undef tenant", func(t *testing.T) { + require.ErrorIs(t, s.Add(t.Context(), did.Undef, "user-1"), store.ErrInvalidArgument) + }) + + t.Run("Add returns ErrInvalidArgument for an empty external ID", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.ErrorIs(t, s.Add(t.Context(), tenantID, ""), store.ErrInvalidArgument) + }) + + t.Run("ListByTenant isolates by tenant and orders by external ID", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + other := testutil.RandomDID(t) + seed(t, tenantID) + seed(t, other) + for _, id := range []string{"c", "a", "b"} { + require.NoError(t, s.Add(t.Context(), tenantID, id)) + } + require.NoError(t, s.Add(t.Context(), other, "a")) + + recs, err := s.ListByTenant(t.Context(), tenantID) + require.NoError(t, err) + require.Len(t, recs, 3) + var ids []string + for _, r := range recs { + require.Equal(t, tenantID, r.Tenant) + ids = append(ids, r.ExternalID) + } + require.Equal(t, []string{"a", "b", "c"}, ids) + + recs, err = s.ListByTenant(t.Context(), testutil.RandomDID(t)) + require.NoError(t, err) + require.Empty(t, recs) + }) + + t.Run("Add revives a removed principal with a fresh CreatedAt", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "again")) + first, err := s.Get(t.Context(), tenantID, "again") + require.NoError(t, err) + require.NoError(t, s.Tombstone(t.Context(), tenantID, "again", nil)) + + recs, err := s.ListByTenant(t.Context(), tenantID) + require.NoError(t, err) + require.Empty(t, recs, "a removed principal is not listed") + + time.Sleep(2 * time.Millisecond) + require.NoError(t, s.Add(t.Context(), tenantID, "again"), "revive is not a duplicate") + revived, err := s.Get(t.Context(), tenantID, "again") + require.NoError(t, err) + require.True(t, revived.CreatedAt.After(first.CreatedAt)) + require.ErrorIs(t, s.Add(t.Context(), tenantID, "again"), store.ErrRecordExists) + }) + + t.Run("Tombstone runs the callback once, removes the row and is idempotent", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "del")) + + calls := 0 + require.NoError(t, s.Tombstone(t.Context(), tenantID, "del", func(context.Context) error { + calls++ + return nil + })) + require.Equal(t, 1, calls) + _, err := s.Get(t.Context(), tenantID, "del") + require.ErrorIs(t, err, store.ErrRecordNotFound) + + // Deleting an absent principal is a no-op: nothing to publish, so the + // callback does not run. + require.NoError(t, s.Tombstone(t.Context(), tenantID, "del", func(context.Context) error { + calls++ + return nil + })) + require.Equal(t, 1, calls) + }) + + t.Run("Tombstone holds only the removal: ListByTenant answers, a share-locked Get waits", func(t *testing.T) { + // The locking both backends owe the two writers that cross here: + // a policy write lists the tenant's principals from inside its own + // lock while a removal's callback rewrites that tenant's policies. + // The list must not wait on the removal, and a share-locked read + // must. + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "held")) + + deleted, got := htestutil.RequireWaitsForWriter(t, + func(entered chan<- struct{}, release <-chan struct{}) error { + return s.Tombstone(context.Background(), tenantID, "held", func(ctx context.Context) error { + recs, err := s.ListByTenant(ctx, tenantID) + if err != nil { + return err + } + if len(recs) != 1 { + return fmt.Errorf("listed %d principals during the removal, want the live one", len(recs)) + } + close(entered) + <-release + return nil + }) + }, + func() error { + _, err := s.Get(context.Background(), tenantID, "held", store.WithShareLock()) + return err + }) + require.NoError(t, deleted) + require.ErrorIs(t, got, store.ErrRecordNotFound, "the tombstone is visible once the removal commits") + }) + + t.Run("Tombstone accepts a nil callback", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "nil-cb")) + require.NoError(t, s.Tombstone(t.Context(), tenantID, "nil-cb", nil)) + _, err := s.Get(t.Context(), tenantID, "nil-cb") + require.ErrorIs(t, err, store.ErrRecordNotFound) + }) + + t.Run("Tombstone returns the callback error and leaves the row", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "keep")) + + publishFailed := errors.New("publish failed") + err := s.Tombstone(t.Context(), tenantID, "keep", func(context.Context) error { + return publishFailed + }) + require.ErrorIs(t, err, publishFailed) + + rec, err := s.Get(t.Context(), tenantID, "keep") + require.NoError(t, err) + require.Equal(t, "keep", rec.ExternalID) + + // A retry after the failure completes the removal. + require.NoError(t, s.Tombstone(t.Context(), tenantID, "keep", nil)) + _, err = s.Get(t.Context(), tenantID, "keep") + require.ErrorIs(t, err, store.ErrRecordNotFound) + }) + + t.Run("WithLock runs the callback once and returns its error", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "held")) + + calls := 0 + require.NoError(t, s.WithLock(t.Context(), tenantID, []string{"held", "absent"}, func(ctx context.Context) error { + calls++ + return nil + })) + require.Equal(t, 1, calls) + + boom := errors.New("boom") + err := s.WithLock(t.Context(), tenantID, []string{"held"}, func(ctx context.Context) error { return boom }) + require.ErrorIs(t, err, boom) + }) + + t.Run("DeleteByTenant removes every principal of the tenant and is idempotent", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + other := testutil.RandomDID(t) + seed(t, tenantID) + seed(t, other) + require.NoError(t, s.Add(t.Context(), tenantID, "a")) + require.NoError(t, s.Add(t.Context(), tenantID, "b")) + require.NoError(t, s.Add(t.Context(), other, "a")) + + require.NoError(t, s.DeleteByTenant(t.Context(), tenantID)) + recs, err := s.ListByTenant(t.Context(), tenantID) + require.NoError(t, err) + require.Empty(t, recs) + recs, err = s.ListByTenant(t.Context(), other) + require.NoError(t, err) + require.Len(t, recs, 1) + + require.NoError(t, s.DeleteByTenant(t.Context(), tenantID)) + }) + }) + } +} + +// TestPrincipalStorePostgresLocking pins the locking contract the authorizer +// relies on: a share-locked Get waits for a transaction that holds the row FOR +// UPDATE, and is answered once that transaction commits. Postgres only: the +// memory store cannot hold a row across calls. +func TestPrincipalStorePostgresLocking(t *testing.T) { + pool := htestutil.PostgresOrSkip(t) + s := principalpostgres.New(pool) + seed := seeder(pool) + + t.Run("a FOR UPDATE holder blocks a share-locked Get until commit", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "locked")) + + tx, err := pool.Begin(t.Context()) + require.NoError(t, err) + defer tx.Rollback(t.Context()) + + var rec principal.Record + committed, got := htestutil.RequireWaitsForWriter(t, + func(entered chan<- struct{}, release <-chan struct{}) error { + var found bool + if err := tx.QueryRow(t.Context(), + `SELECT TRUE FROM principal WHERE tenant_id = $1 AND external_id = $2 FOR UPDATE`, + tenantID.String(), "locked").Scan(&found); err != nil { + return err + } + // An unlocked read is answered from the snapshot and never waits. + if _, err := s.Get(t.Context(), tenantID, "locked"); err != nil { + return fmt.Errorf("unlocked Get during the hold: %w", err) + } + close(entered) + <-release + return tx.Commit(t.Context()) + }, + func() error { + var err error + rec, err = s.Get(context.Background(), tenantID, "locked", store.WithShareLock()) + return err + }) + require.NoError(t, committed) + require.NoError(t, got) + require.Equal(t, "locked", rec.ExternalID) + }) + + t.Run("a share-locked Get waits for WithLock's callback", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "held")) + + locked, got := htestutil.RequireWaitsForWriter(t, + func(entered chan<- struct{}, release <-chan struct{}) error { + return s.WithLock(context.Background(), tenantID, []string{"held"}, func(ctx context.Context) error { + close(entered) + <-release + return nil + }) + }, + func() error { + _, err := s.Get(context.Background(), tenantID, "held", store.WithShareLock()) + return err + }) + require.NoError(t, locked) + require.NoError(t, got) + }) + + t.Run("a share-locked Get during Tombstone sees the row once the callback fails", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "rollback")) + + var rec principal.Record + deleted, got := htestutil.RequireWaitsForWriter(t, + func(entered chan<- struct{}, release <-chan struct{}) error { + return s.Tombstone(context.Background(), tenantID, "rollback", func(context.Context) error { + close(entered) + <-release + return errors.New("publish failed") + }) + }, + func() error { + var err error + rec, err = s.Get(context.Background(), tenantID, "rollback", store.WithShareLock()) + return err + }) + require.Error(t, deleted) + require.NoError(t, got, "the rolled-back delete must leave the row readable") + require.Equal(t, "rollback", rec.ExternalID) + }) + + t.Run("a share-locked Get during Tombstone reports the row gone once it commits", func(t *testing.T) { + tenantID := testutil.RandomDID(t) + seed(t, tenantID) + require.NoError(t, s.Add(t.Context(), tenantID, "gone")) + + deleted, got := htestutil.RequireWaitsForWriter(t, + func(entered chan<- struct{}, release <-chan struct{}) error { + return s.Tombstone(context.Background(), tenantID, "gone", func(context.Context) error { + close(entered) + <-release + return nil + }) + }, + func() error { + _, err := s.Get(context.Background(), tenantID, "gone", store.WithShareLock()) + return err + }) + require.NoError(t, deleted) + require.ErrorIs(t, got, store.ErrRecordNotFound) + }) +} diff --git a/pkg/store/store.go b/pkg/store/store.go index 657a0931..d431109b 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -55,3 +55,35 @@ func Collect[T any](ctx context.Context, getPage GetPageFunc[T]) ([]T, error) { } return items, nil } + +// ReadConfig is the row lock a read takes. The zero value takes none. +type ReadConfig struct { + // Share reads the row with SELECT ... FOR SHARE. A write in flight on the + // same row (held FOR UPDATE inside its transaction) blocks the read until it + // commits or rolls back, so the read is answered from the committed state. + // The read needs no transaction of its own: a FOR SHARE in autocommit mode + // still waits on a conflicting FOR UPDATE. Without it a read never waits and + // may be answered from a snapshot that predates an in-flight write. + // + // The memory backends give the same guarantee with their own locks: a + // share-locked read there waits for a write in flight on the same record. + Share bool +} + +type ReadOption func(cfg *ReadConfig) + +// WithShareLock reads with [ReadConfig.Share]. +func WithShareLock() ReadOption { + return func(cfg *ReadConfig) { + cfg.Share = true + } +} + +// NewReadConfig applies opts to the zero ReadConfig. +func NewReadConfig(opts ...ReadOption) ReadConfig { + var cfg ReadConfig + for _, opt := range opts { + opt(&cfg) + } + return cfg +}