From 6c17ff8bea95b9c9cd4440e423fddabd56ff1821 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Thu, 10 Sep 2026 18:09:26 +0200 Subject: [PATCH 01/11] feat(store): add the principal store and share-locked reads Adds pkg/store/principal with memory and Postgres backends behind one suite, and migration 00006_principal.sql with the RFC's principal table. Delete locks the row FOR UPDATE, runs a caller-supplied beforeCommit callback while holding it, then deletes and commits. A callback error rolls the transaction back and leaves the row in place; deleting an absent row is a no-op that skips the callback. Adds store.ReadOption and store.WithLock(store.LockShare). The Postgres Get honours it as SELECT ... FOR SHARE, so a read waits for an in-flight Delete of the same row to commit or roll back. The memory backend already serializes under its mutex and ignores the option. Wires the store into both fx store modules and has tenant deletion remove the tenant's principals before the tenant row, which they reference with ON DELETE RESTRICT. Co-Authored-By: Claude Fable 5.1 --- internal/testutil/postgres.go | 15 + internal/testutil/wait.go | 40 +++ pkg/api/service/tenant/service.go | 15 +- pkg/api/service/tenant/service_test.go | 22 +- pkg/api/tenants_test.go | 17 +- pkg/fx/store/memory/provider.go | 3 + pkg/fx/store/postgres/provider.go | 7 + pkg/migrations/sql/00006_principal.sql | 20 ++ pkg/store/principal/memory/store.go | 172 +++++++++++ pkg/store/principal/postgres/store.go | 200 ++++++++++++ pkg/store/principal/principal.go | 66 ++++ pkg/store/principal/principal_test.go | 408 +++++++++++++++++++++++++ pkg/store/store.go | 14 + 13 files changed, 986 insertions(+), 13 deletions(-) create mode 100644 internal/testutil/wait.go create mode 100644 pkg/migrations/sql/00006_principal.sql create mode 100644 pkg/store/principal/memory/store.go create mode 100644 pkg/store/principal/postgres/store.go create mode 100644 pkg/store/principal/principal.go create mode 100644 pkg/store/principal/principal_test.go 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..a500445d --- /dev/null +++ b/internal/testutil/wait.go @@ -0,0 +1,40 @@ +package testutil + +import ( + "testing" + "time" +) + +// Grace is how long a call that must wait on a writer is watched for +// returning early. +const Grace = 300 * time.Millisecond + +// RequireWaitsForWriter runs write in the background; write must close +// entered once it holds its lock and park on release before finishing. It +// then starts wait, requires that wait has not returned within [Grace], +// 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) }() + <-entered + + 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(Grace): + } + + close(release) + writeErr = <-wrote + select { + case waitErr = <-waited: + case <-time.After(10 * time.Second): + 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..4f5f237c --- /dev/null +++ b/pkg/store/principal/memory/store.go @@ -0,0 +1,172 @@ +// 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 +// Delete holds while beforeCommit 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.LockShare] waits for an in-flight Delete 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, locks ...store.LockMode) (principal.Record, error) { + if slices.Contains(locks, store.LockShare) { + 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 +} + +// Delete holds removals for the whole call and takes the map mutex only to +// read the entry and, at the end, to write the tombstone. beforeCommit 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) Delete(ctx context.Context, tenant did.DID, externalID string, beforeCommit 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 beforeCommit != nil { + if err := beforeCommit(ctx); err != nil { + return fmt.Errorf("before deleting 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 +} + +// Lock runs fn without holding anything: the exclusion it provides on +// Postgres is a row lock a concurrent removal or share-locked read waits on, +// and the memory backend has no waiting reader to serve. +func (s *Store) Lock(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..2c65ea16 --- /dev/null +++ b/pkg/store/principal/postgres/store.go @@ -0,0 +1,200 @@ +// Package postgres provides a PostgreSQL-backed implementation of +// principal.Store. +package postgres + +import ( + "context" + "errors" + "fmt" + "slices" + "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 Delete 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.LockShare] is requested so the +// read waits on a Delete that holds the row FOR UPDATE. +func (s *Store) Get(ctx context.Context, tenant did.DID, externalID string, locks ...store.LockMode) (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 slices.Contains(locks, store.LockShare) { + 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 +} + +// Delete 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) Delete(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 deleting 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("deleting principal: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } + return nil +} + +// Lock 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 Lock +// calls over overlapping principals from deadlocking. +func (s *Store) Lock(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..48c0e0d8 --- /dev/null +++ b/pkg/store/principal/principal.go @@ -0,0 +1,66 @@ +// 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.LockShare] the read waits for an in-flight [Store.Delete] of the + // same row to commit or roll back. + Get(ctx context.Context, tenant did.DID, externalID string, locks ...store.LockMode) (Record, error) + // ListByTenant returns every live principal of the tenant, ordered by + // external ID. + ListByTenant(ctx context.Context, tenant did.DID) ([]Record, error) + // Delete removes a principal by marking its row deleted. It excludes + // concurrent removals and revives of the same row for the whole call, runs + // beforeCommit (nil allowed), then marks the row and commits. An error from + // beforeCommit is returned and leaves the principal live. It is idempotent: + // when no live row exists it returns nil without running beforeCommit. + // + // beforeCommit 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. + Delete(ctx context.Context, tenant did.DID, externalID string, beforeCommit func(ctx context.Context) error) error + // Lock holds the tenant's live principals with the given external IDs + // against concurrent removals, revives and share-locked reads while fn runs, + // then releases them. 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.Delete]'s beforeCommit. + Lock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error + // DeleteByTenant deletes every row 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..281d3079 --- /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.LockShare) + 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.LockShare) + 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.Delete(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("Delete 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.Delete(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.Delete(t.Context(), tenantID, "del", func(context.Context) error { + calls++ + return nil + })) + require.Equal(t, 1, calls) + }) + + t.Run("Delete 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.Delete(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.LockShare) + return err + }) + require.NoError(t, deleted) + require.ErrorIs(t, got, store.ErrRecordNotFound, "the tombstone is visible once the removal commits") + }) + + t.Run("Delete 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.Delete(t.Context(), tenantID, "nil-cb", nil)) + _, err := s.Get(t.Context(), tenantID, "nil-cb") + require.ErrorIs(t, err, store.ErrRecordNotFound) + }) + + t.Run("Delete 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.Delete(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.Delete(t.Context(), tenantID, "keep", nil)) + _, err = s.Get(t.Context(), tenantID, "keep") + require.ErrorIs(t, err, store.ErrRecordNotFound) + }) + + t.Run("Lock 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.Lock(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.Lock(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.LockShare) + return err + }) + require.NoError(t, committed) + require.NoError(t, got) + require.Equal(t, "locked", rec.ExternalID) + }) + + t.Run("a share-locked Get waits for Lock'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.Lock(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.LockShare) + return err + }) + require.NoError(t, locked) + require.NoError(t, got) + }) + + t.Run("a share-locked Get during Delete 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.Delete(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.LockShare) + 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 Delete 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.Delete(context.Background(), tenantID, "gone", func(context.Context) error { + close(entered) + <-release + return nil + }) + }, + func() error { + _, err := s.Get(context.Background(), tenantID, "gone", store.LockShare) + 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..a0632ac7 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -55,3 +55,17 @@ func Collect[T any](ctx context.Context, getPage GetPageFunc[T]) ([]T, error) { } return items, nil } + +// LockMode is the row lock a read takes. +type LockMode int + +// LockShare 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. +const LockShare LockMode = iota + 1 From bbafa88dbec2d2ba24a4489f79d4e94693ee962f Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Fri, 25 Sep 2026 17:56:34 +0200 Subject: [PATCH 02/11] docs(store): say why the memory Lock holds nothing The comment claimed the memory backend "has no waiting reader to serve", which is wrong: a share-locked Get is exactly that reader, and it does wait on a removal. The real reason is the deadlock. A policy write calls Lock while holding the bucket's policy, and a removal holds removals while its callback rewrites that same policy, so each would wait on what the other holds. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/store/principal/memory/store.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/pkg/store/principal/memory/store.go b/pkg/store/principal/memory/store.go index 4f5f237c..24e6ee9b 100644 --- a/pkg/store/principal/memory/store.go +++ b/pkg/store/principal/memory/store.go @@ -152,9 +152,11 @@ func (s *Store) Delete(ctx context.Context, tenant did.DID, externalID string, b return nil } -// Lock runs fn without holding anything: the exclusion it provides on -// Postgres is a row lock a concurrent removal or share-locked read waits on, -// and the memory backend has no waiting reader to serve. +// Lock 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 Lock 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) Lock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error { return fn(ctx) } From 8f8bb3e5c871b3f4dac8156de717bbb5b8210e35 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Wed, 30 Sep 2026 17:24:29 +0200 Subject: [PATCH 03/11] refactor(store): take the read lock as a functional option store.WithShareLock replaces the LockMode variadic, following the PaginationOption pattern. A read with no option takes no lock. Co-Authored-By: Claude Opus 5.5 --- pkg/store/principal/memory/store.go | 6 ++-- pkg/store/principal/postgres/store.go | 7 ++--- pkg/store/principal/principal.go | 4 +-- pkg/store/principal/principal_test.go | 14 ++++----- pkg/store/store.go | 42 +++++++++++++++++++-------- 5 files changed, 45 insertions(+), 28 deletions(-) diff --git a/pkg/store/principal/memory/store.go b/pkg/store/principal/memory/store.go index 24e6ee9b..5467412e 100644 --- a/pkg/store/principal/memory/store.go +++ b/pkg/store/principal/memory/store.go @@ -76,13 +76,13 @@ func (s *Store) Add(ctx context.Context, tenant did.DID, externalID string) erro return nil } -// Get with [store.LockShare] waits for an in-flight Delete of any principal +// Get with [store.WithShareLock] waits for an in-flight Delete 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, locks ...store.LockMode) (principal.Record, error) { - if slices.Contains(locks, store.LockShare) { +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() } diff --git a/pkg/store/principal/postgres/store.go b/pkg/store/principal/postgres/store.go index 2c65ea16..f4467076 100644 --- a/pkg/store/principal/postgres/store.go +++ b/pkg/store/principal/postgres/store.go @@ -6,7 +6,6 @@ import ( "context" "errors" "fmt" - "slices" "time" "github.com/fil-forge/hilt/pkg/store" @@ -55,15 +54,15 @@ func (s *Store) Add(ctx context.Context, tenant did.DID, externalID string) erro return nil } -// Get reads the row, with FOR SHARE when [store.LockShare] is requested so the +// Get reads the row, with FOR SHARE when [store.WithShareLock] is passed so the // read waits on a Delete that holds the row FOR UPDATE. -func (s *Store) Get(ctx context.Context, tenant did.DID, externalID string, locks ...store.LockMode) (principal.Record, error) { +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 slices.Contains(locks, store.LockShare) { + if store.NewReadConfig(opts...).Share { query += ` FOR SHARE` } rec, err := scanRecord(s.pool.QueryRow(ctx, query, tenant.String(), externalID)) diff --git a/pkg/store/principal/principal.go b/pkg/store/principal/principal.go index 48c0e0d8..bc1f9ed4 100644 --- a/pkg/store/principal/principal.go +++ b/pkg/store/principal/principal.go @@ -36,9 +36,9 @@ type Store interface { 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.LockShare] the read waits for an in-flight [Store.Delete] of the + // [store.WithShareLock] the read waits for an in-flight [Store.Delete] of the // same row to commit or roll back. - Get(ctx context.Context, tenant did.DID, externalID string, locks ...store.LockMode) (Record, error) + 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) diff --git a/pkg/store/principal/principal_test.go b/pkg/store/principal/principal_test.go index 281d3079..e01bfb4f 100644 --- a/pkg/store/principal/principal_test.go +++ b/pkg/store/principal/principal_test.go @@ -78,7 +78,7 @@ func TestPrincipalStore(t *testing.T) { seed(t, tenantID) require.NoError(t, s.Add(t.Context(), tenantID, "user-1")) - rec, err := s.Get(t.Context(), tenantID, "user-1", store.LockShare) + rec, err := s.Get(t.Context(), tenantID, "user-1", store.WithShareLock()) require.NoError(t, err) require.Equal(t, "user-1", rec.ExternalID) }) @@ -88,7 +88,7 @@ func TestPrincipalStore(t *testing.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.LockShare) + _, err = s.Get(t.Context(), tenantID, "nobody", store.WithShareLock()) require.ErrorIs(t, err, store.ErrRecordNotFound) }) @@ -218,7 +218,7 @@ func TestPrincipalStore(t *testing.T) { }) }, func() error { - _, err := s.Get(context.Background(), tenantID, "held", store.LockShare) + _, err := s.Get(context.Background(), tenantID, "held", store.WithShareLock()) return err }) require.NoError(t, deleted) @@ -332,7 +332,7 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { }, func() error { var err error - rec, err = s.Get(context.Background(), tenantID, "locked", store.LockShare) + rec, err = s.Get(context.Background(), tenantID, "locked", store.WithShareLock()) return err }) require.NoError(t, committed) @@ -354,7 +354,7 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { }) }, func() error { - _, err := s.Get(context.Background(), tenantID, "held", store.LockShare) + _, err := s.Get(context.Background(), tenantID, "held", store.WithShareLock()) return err }) require.NoError(t, locked) @@ -377,7 +377,7 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { }, func() error { var err error - rec, err = s.Get(context.Background(), tenantID, "rollback", store.LockShare) + rec, err = s.Get(context.Background(), tenantID, "rollback", store.WithShareLock()) return err }) require.Error(t, deleted) @@ -399,7 +399,7 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { }) }, func() error { - _, err := s.Get(context.Background(), tenantID, "gone", store.LockShare) + _, err := s.Get(context.Background(), tenantID, "gone", store.WithShareLock()) return err }) require.NoError(t, deleted) diff --git a/pkg/store/store.go b/pkg/store/store.go index a0632ac7..d431109b 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -56,16 +56,34 @@ func Collect[T any](ctx context.Context, getPage GetPageFunc[T]) ([]T, error) { return items, nil } -// LockMode is the row lock a read takes. -type LockMode int +// 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) -// LockShare 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. -const LockShare LockMode = iota + 1 +// 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 +} From 6c264b0c23d4057f326cb0a44265ba077d7c8136 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Wed, 30 Sep 2026 18:12:05 +0200 Subject: [PATCH 04/11] test: fail instead of hanging when the writer exits before its lock Co-Authored-By: Claude Opus 5.5 --- internal/testutil/wait.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/internal/testutil/wait.go b/internal/testutil/wait.go index a500445d..94c79e25 100644 --- a/internal/testutil/wait.go +++ b/internal/testutil/wait.go @@ -19,7 +19,13 @@ func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, rel release := make(chan struct{}) wrote := make(chan error, 1) go func() { wrote <- write(entered, release) }() - <-entered + select { + case <-entered: + case err := <-wrote: + t.Fatalf("the writer returned before taking its lock: %v", err) + case <-time.After(10 * time.Second): + t.Fatal("the writer did not take its lock") + } waited := make(chan error, 1) go func() { waited <- wait() }() From d7624e5b3299e81a022f122baaff7f6d2daff6b4 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Wed, 30 Sep 2026 18:13:47 +0200 Subject: [PATCH 05/11] test: name the wait helper's timeout Co-Authored-By: Claude Opus 5.5 --- internal/testutil/wait.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/internal/testutil/wait.go b/internal/testutil/wait.go index 94c79e25..c279a3fa 100644 --- a/internal/testutil/wait.go +++ b/internal/testutil/wait.go @@ -9,6 +9,9 @@ import ( // returning early. const Grace = 300 * time.Millisecond +// Patience is how long a call that must finish is given before the test fails. +const Patience = 10 * time.Second + // RequireWaitsForWriter runs write in the background; write must close // entered once it holds its lock and park on release before finishing. It // then starts wait, requires that wait has not returned within [Grace], @@ -23,7 +26,7 @@ func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, rel case <-entered: case err := <-wrote: t.Fatalf("the writer returned before taking its lock: %v", err) - case <-time.After(10 * time.Second): + case <-time.After(Patience): t.Fatal("the writer did not take its lock") } @@ -39,7 +42,7 @@ func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, rel writeErr = <-wrote select { case waitErr = <-waited: - case <-time.After(10 * time.Second): + case <-time.After(Patience): t.Fatal("the waiting call did not return after the writer finished") } return writeErr, waitErr From 32070edd5cc87146ba6922b5a3a88d401ee0c46c Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Wed, 30 Sep 2026 18:22:34 +0200 Subject: [PATCH 06/11] test: call the wait helper's timeout WaitTimeout Co-Authored-By: Claude Opus 5.5 --- internal/testutil/wait.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/testutil/wait.go b/internal/testutil/wait.go index c279a3fa..8e6ad592 100644 --- a/internal/testutil/wait.go +++ b/internal/testutil/wait.go @@ -9,8 +9,8 @@ import ( // returning early. const Grace = 300 * time.Millisecond -// Patience is how long a call that must finish is given before the test fails. -const Patience = 10 * time.Second +// 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 and park on release before finishing. It @@ -26,7 +26,7 @@ func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, rel case <-entered: case err := <-wrote: t.Fatalf("the writer returned before taking its lock: %v", err) - case <-time.After(Patience): + case <-time.After(WaitTimeout): t.Fatal("the writer did not take its lock") } @@ -42,7 +42,7 @@ func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, rel writeErr = <-wrote select { case waitErr = <-waited: - case <-time.After(Patience): + case <-time.After(WaitTimeout): t.Fatal("the waiting call did not return after the writer finished") } return writeErr, waitErr From 8f1a8d06fe4b2b0bad249711b6b60339b4a13d87 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Wed, 30 Sep 2026 18:57:47 +0200 Subject: [PATCH 07/11] test: call the wait helper's blocked window MinBlock Co-Authored-By: Claude Opus 5.5 --- internal/testutil/wait.go | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/internal/testutil/wait.go b/internal/testutil/wait.go index 8e6ad592..0c803cd9 100644 --- a/internal/testutil/wait.go +++ b/internal/testutil/wait.go @@ -5,16 +5,15 @@ import ( "time" ) -// Grace is how long a call that must wait on a writer is watched for -// returning early. -const Grace = 300 * time.Millisecond +// 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 and park on release before finishing. It -// then starts wait, requires that wait has not returned within [Grace], +// 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() @@ -35,7 +34,7 @@ func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, rel select { case <-waited: t.Fatal("the waiting call returned while the writer held its lock") - case <-time.After(Grace): + case <-time.After(MinBlock): } close(release) From 2e6ad4797995f8c7ba71adc26306a9f83918f61d Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Mon, 5 Oct 2026 15:51:32 +0200 Subject: [PATCH 08/11] docs(testutil): say that the writer blocks until release is closed Co-Authored-By: Claude Fable 5.1 --- internal/testutil/wait.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/testutil/wait.go b/internal/testutil/wait.go index 0c803cd9..b6805e5e 100644 --- a/internal/testutil/wait.go +++ b/internal/testutil/wait.go @@ -12,8 +12,8 @@ const MinBlock = 300 * time.Millisecond const WaitTimeout = 10 * time.Second // RequireWaitsForWriter runs write in the background; write must close -// entered once it holds its lock and park on release before finishing. It -// then starts wait, requires that wait has not returned within [MinBlock], +// 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() From 08ee0633d45fd29e21202e1a579ae3d4713ddb31 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Mon, 5 Oct 2026 15:53:18 +0200 Subject: [PATCH 09/11] refactor(store): call the principal removal Tombstone and its callback fn Store.Delete marked the row removed while DeleteByTenant deletes rows, so the two read as the same operation. Tombstone says what the write does. The callback is fn: "before commit" names a transaction, and the memory backend has none. The contract is unchanged: fn runs inside the write, before it takes effect, and its error leaves the principal live. Co-Authored-By: Claude Fable 5.1 --- pkg/store/principal/memory/store.go | 18 ++++++++-------- pkg/store/principal/postgres/store.go | 18 ++++++++-------- pkg/store/principal/principal.go | 31 ++++++++++++++------------- pkg/store/principal/principal_test.go | 30 +++++++++++++------------- 4 files changed, 49 insertions(+), 48 deletions(-) diff --git a/pkg/store/principal/memory/store.go b/pkg/store/principal/memory/store.go index 5467412e..472eb90c 100644 --- a/pkg/store/principal/memory/store.go +++ b/pkg/store/principal/memory/store.go @@ -3,7 +3,7 @@ // 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 -// Delete holds while beforeCommit runs. +// 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 @@ -76,7 +76,7 @@ func (s *Store) Add(ctx context.Context, tenant did.DID, externalID string) erro return nil } -// Get with [store.WithShareLock] waits for an in-flight Delete of any principal +// 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 @@ -113,13 +113,13 @@ func (s *Store) ListByTenant(ctx context.Context, tenant did.DID) ([]principal.R return recs, nil } -// Delete holds removals for the whole call and takes the map mutex only to -// read the entry and, at the end, to write the tombstone. beforeCommit runs -// with the map unlocked, so it may write the policy store whose own writes +// 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) Delete(ctx context.Context, tenant did.DID, externalID string, beforeCommit func(ctx context.Context) error) error { +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() @@ -130,9 +130,9 @@ func (s *Store) Delete(ctx context.Context, tenant did.DID, externalID string, b return nil // idempotent: nothing to publish and nothing to remove } - if beforeCommit != nil { - if err := beforeCommit(ctx); err != nil { - return fmt.Errorf("before deleting principal: %w", err) + if fn != nil { + if err := fn(ctx); err != nil { + return fmt.Errorf("before tombstoning principal: %w", err) } } diff --git a/pkg/store/principal/postgres/store.go b/pkg/store/principal/postgres/store.go index f4467076..975eba3b 100644 --- a/pkg/store/principal/postgres/store.go +++ b/pkg/store/principal/postgres/store.go @@ -37,7 +37,7 @@ func (s *Store) Add(ctx context.Context, tenant did.DID, externalID string) erro } // 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 Delete of the same principal. + // 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) @@ -55,7 +55,7 @@ func (s *Store) Add(ctx context.Context, tenant did.DID, externalID string) erro } // Get reads the row, with FOR SHARE when [store.WithShareLock] is passed so the -// read waits on a Delete that holds the row FOR UPDATE. +// 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 @@ -101,10 +101,10 @@ func (s *Store) ListByTenant(ctx context.Context, tenant did.DID) ([]principal.R return recs, nil } -// Delete runs in one transaction: it locks the row FOR UPDATE, runs -// beforeCommit while holding the lock, sets deleted_at and commits. A locked +// Tombstone runs in one transaction: it locks the row FOR UPDATE, runs fn +// 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) Delete(ctx context.Context, tenant did.DID, externalID string, beforeCommit func(ctx context.Context) error) error { +func (s *Store) Tombstone(ctx context.Context, tenant did.DID, externalID string, fn func(ctx context.Context) error) error { tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("beginning transaction: %w", err) @@ -125,9 +125,9 @@ func (s *Store) Delete(ctx context.Context, tenant did.DID, externalID string, b return fmt.Errorf("locking principal: %w", err) } - if beforeCommit != nil { - if err := beforeCommit(ctx); err != nil { - return fmt.Errorf("before deleting principal: %w", err) + if fn != nil { + if err := fn(ctx); err != nil { + return fmt.Errorf("before tombstoning principal: %w", err) } } @@ -136,7 +136,7 @@ func (s *Store) Delete(ctx context.Context, tenant did.DID, externalID string, b SET deleted_at = NOW() WHERE tenant_id = $1 AND external_id = $2 `, tenant.String(), externalID); err != nil { - return fmt.Errorf("deleting principal: %w", err) + return fmt.Errorf("tombstoning principal: %w", err) } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("committing transaction: %w", err) diff --git a/pkg/store/principal/principal.go b/pkg/store/principal/principal.go index bc1f9ed4..b365e945 100644 --- a/pkg/store/principal/principal.go +++ b/pkg/store/principal/principal.go @@ -36,31 +36,32 @@ type Store interface { 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.Delete] of the - // same row to commit or roll back. + // [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) - // Delete removes a principal by marking its row deleted. It excludes - // concurrent removals and revives of the same row for the whole call, runs - // beforeCommit (nil allowed), then marks the row and commits. An error from - // beforeCommit is returned and leaves the principal live. It is idempotent: - // when no live row exists it returns nil without running beforeCommit. + // 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. // - // beforeCommit 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. - Delete(ctx context.Context, tenant did.DID, externalID string, beforeCommit func(ctx context.Context) error) error + // 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 // Lock holds the tenant's live principals with the given external IDs // against concurrent removals, revives and share-locked reads while fn runs, // then releases them. 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.Delete]'s beforeCommit. + // follows the contract of [Store.Tombstone]'s fn. Lock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error - // DeleteByTenant deletes every row of the tenant, tombstones included. It is - // idempotent. + // 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 index e01bfb4f..9cda4e34 100644 --- a/pkg/store/principal/principal_test.go +++ b/pkg/store/principal/principal_test.go @@ -155,7 +155,7 @@ func TestPrincipalStore(t *testing.T) { 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.Delete(t.Context(), tenantID, "again", nil)) + require.NoError(t, s.Tombstone(t.Context(), tenantID, "again", nil)) recs, err := s.ListByTenant(t.Context(), tenantID) require.NoError(t, err) @@ -169,13 +169,13 @@ func TestPrincipalStore(t *testing.T) { require.ErrorIs(t, s.Add(t.Context(), tenantID, "again"), store.ErrRecordExists) }) - t.Run("Delete runs the callback once, removes the row and is idempotent", func(t *testing.T) { + 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.Delete(t.Context(), tenantID, "del", func(context.Context) error { + require.NoError(t, s.Tombstone(t.Context(), tenantID, "del", func(context.Context) error { calls++ return nil })) @@ -185,14 +185,14 @@ func TestPrincipalStore(t *testing.T) { // Deleting an absent principal is a no-op: nothing to publish, so the // callback does not run. - require.NoError(t, s.Delete(t.Context(), tenantID, "del", func(context.Context) error { + require.NoError(t, s.Tombstone(t.Context(), tenantID, "del", func(context.Context) error { calls++ return nil })) require.Equal(t, 1, calls) }) - t.Run("Delete holds only the removal: ListByTenant answers, a share-locked Get waits", func(t *testing.T) { + 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. @@ -204,7 +204,7 @@ func TestPrincipalStore(t *testing.T) { deleted, got := htestutil.RequireWaitsForWriter(t, func(entered chan<- struct{}, release <-chan struct{}) error { - return s.Delete(context.Background(), tenantID, "held", func(ctx context.Context) error { + return s.Tombstone(context.Background(), tenantID, "held", func(ctx context.Context) error { recs, err := s.ListByTenant(ctx, tenantID) if err != nil { return err @@ -225,22 +225,22 @@ func TestPrincipalStore(t *testing.T) { require.ErrorIs(t, got, store.ErrRecordNotFound, "the tombstone is visible once the removal commits") }) - t.Run("Delete accepts a nil callback", func(t *testing.T) { + 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.Delete(t.Context(), tenantID, "nil-cb", nil)) + 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("Delete returns the callback error and leaves the row", func(t *testing.T) { + 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.Delete(t.Context(), tenantID, "keep", func(context.Context) error { + err := s.Tombstone(t.Context(), tenantID, "keep", func(context.Context) error { return publishFailed }) require.ErrorIs(t, err, publishFailed) @@ -250,7 +250,7 @@ func TestPrincipalStore(t *testing.T) { require.Equal(t, "keep", rec.ExternalID) // A retry after the failure completes the removal. - require.NoError(t, s.Delete(t.Context(), tenantID, "keep", nil)) + require.NoError(t, s.Tombstone(t.Context(), tenantID, "keep", nil)) _, err = s.Get(t.Context(), tenantID, "keep") require.ErrorIs(t, err, store.ErrRecordNotFound) }) @@ -361,7 +361,7 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { require.NoError(t, got) }) - t.Run("a share-locked Get during Delete sees the row once the callback fails", func(t *testing.T) { + 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")) @@ -369,7 +369,7 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { var rec principal.Record deleted, got := htestutil.RequireWaitsForWriter(t, func(entered chan<- struct{}, release <-chan struct{}) error { - return s.Delete(context.Background(), tenantID, "rollback", func(context.Context) error { + return s.Tombstone(context.Background(), tenantID, "rollback", func(context.Context) error { close(entered) <-release return errors.New("publish failed") @@ -385,14 +385,14 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { require.Equal(t, "rollback", rec.ExternalID) }) - t.Run("a share-locked Get during Delete reports the row gone once it commits", func(t *testing.T) { + 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.Delete(context.Background(), tenantID, "gone", func(context.Context) error { + return s.Tombstone(context.Background(), tenantID, "gone", func(context.Context) error { close(entered) <-release return nil From 56ef319cb0d17a5bd8246c4ae749fd0f5c2eb2b2 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Mon, 5 Oct 2026 15:54:00 +0200 Subject: [PATCH 10/11] refactor(store): rename the principal store's Lock to WithLock Lock read as taking a lock the caller gives back. WithLock says the lock is held for fn and released when fn returns. Co-Authored-By: Claude Fable 5.1 --- pkg/store/principal/memory/store.go | 6 +++--- pkg/store/principal/postgres/store.go | 8 ++++---- pkg/store/principal/principal.go | 8 ++++---- pkg/store/principal/principal_test.go | 10 +++++----- 4 files changed, 16 insertions(+), 16 deletions(-) diff --git a/pkg/store/principal/memory/store.go b/pkg/store/principal/memory/store.go index 472eb90c..774eb455 100644 --- a/pkg/store/principal/memory/store.go +++ b/pkg/store/principal/memory/store.go @@ -152,12 +152,12 @@ func (s *Store) Tombstone(ctx context.Context, tenant did.DID, externalID string return nil } -// Lock runs fn without holding anything, and a removal of one of the named +// 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 Lock holds the bucket's policy, and a removal +// 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) Lock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error { +func (s *Store) WithLock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error { return fn(ctx) } diff --git a/pkg/store/principal/postgres/store.go b/pkg/store/principal/postgres/store.go index 975eba3b..f7b99895 100644 --- a/pkg/store/principal/postgres/store.go +++ b/pkg/store/principal/postgres/store.go @@ -144,10 +144,10 @@ func (s *Store) Tombstone(ctx context.Context, tenant did.DID, externalID string return nil } -// Lock 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 Lock -// calls over overlapping principals from deadlocking. -func (s *Store) Lock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error { +// 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) } diff --git a/pkg/store/principal/principal.go b/pkg/store/principal/principal.go index b365e945..d5e0a2c1 100644 --- a/pkg/store/principal/principal.go +++ b/pkg/store/principal/principal.go @@ -54,13 +54,13 @@ type Store interface { // 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 - // Lock holds the tenant's live principals with the given external IDs - // against concurrent removals, revives and share-locked reads while fn runs, - // then releases them. A policy write uses it so that a key created for one + // 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. - Lock(ctx context.Context, tenant did.DID, externalIDs []string, fn func(ctx context.Context) error) error + 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 index 9cda4e34..bf8fbc7f 100644 --- a/pkg/store/principal/principal_test.go +++ b/pkg/store/principal/principal_test.go @@ -255,20 +255,20 @@ func TestPrincipalStore(t *testing.T) { require.ErrorIs(t, err, store.ErrRecordNotFound) }) - t.Run("Lock runs the callback once and returns its error", func(t *testing.T) { + 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.Lock(t.Context(), tenantID, []string{"held", "absent"}, func(ctx context.Context) error { + 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.Lock(t.Context(), tenantID, []string{"held"}, func(ctx context.Context) error { return boom }) + err := s.WithLock(t.Context(), tenantID, []string{"held"}, func(ctx context.Context) error { return boom }) require.ErrorIs(t, err, boom) }) @@ -340,14 +340,14 @@ func TestPrincipalStorePostgresLocking(t *testing.T) { require.Equal(t, "locked", rec.ExternalID) }) - t.Run("a share-locked Get waits for Lock's callback", func(t *testing.T) { + 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.Lock(context.Background(), tenantID, []string{"held"}, func(ctx context.Context) error { + return s.WithLock(context.Background(), tenantID, []string{"held"}, func(ctx context.Context) error { close(entered) <-release return nil From 767c2b2bc972459dc7e10f95ab96e8da55f80df9 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Mon, 5 Oct 2026 16:56:52 +0200 Subject: [PATCH 11/11] refactor(store): keep beforeCommit as the Postgres principal store's callback name The memory store calls it fn; on Postgres the callback does run before the commit. Co-Authored-By: Claude Opus 5.5 --- pkg/store/principal/postgres/store.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pkg/store/principal/postgres/store.go b/pkg/store/principal/postgres/store.go index f7b99895..2d455f4e 100644 --- a/pkg/store/principal/postgres/store.go +++ b/pkg/store/principal/postgres/store.go @@ -101,10 +101,10 @@ func (s *Store) ListByTenant(ctx context.Context, tenant did.DID) ([]principal.R return recs, nil } -// Tombstone runs in one transaction: it locks the row FOR UPDATE, runs fn +// 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, fn func(ctx context.Context) error) error { +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) @@ -125,8 +125,8 @@ func (s *Store) Tombstone(ctx context.Context, tenant did.DID, externalID string return fmt.Errorf("locking principal: %w", err) } - if fn != nil { - if err := fn(ctx); err != nil { + if beforeCommit != nil { + if err := beforeCommit(ctx); err != nil { return fmt.Errorf("before tombstoning principal: %w", err) } }