feat: export sessions pin a tenant's buckets during a credible exit

AGENTS.md

@@ -105,8 +105,8 @@ and `sprue` (the upload service; mirror its patterns where relevant).
 - `pkg/grant` — an access key's stored delegations: `Issue` builds the
   tenant→key set both key kinds hold, `Rotator` rewrites a principal-bound
   key's set as its policies change. See the package doc.
-- `pkg/store/{tenant,accesskey,bucket,delegation,provider,principal,bucketpolicy}` —
-  each an interface with `memory` and `postgres` backends.
+- `pkg/store/{tenant,accesskey,bucket,delegation,provider,principal,bucketpolicy,exportsession}`
+  — each an interface with `memory` and `postgres` backends.
 - `pkg/store/pglock` — the Postgres locking helpers the stores share: bounded
   lock waits, advisory locks, and the mapping to `store.ErrLockTimeout`.
 - `pkg/vault` (`memory`, `openbao`) — private-key storage; `paths.go` has the

pkg/api/service/tenant/errors.go

@@ -14,6 +14,7 @@ const (
 	DIDRegistrationErrorName    = "DIDRegistration"
 	UploadRegistrationErrorName = "UploadRegistration"
 	DIDDeactivationErrorName    = "DIDDeactivation"
+	ExportInProgressErrorName   = "ExportInProgress"
 )
 
 // Known errors returned by the tenant [Service]. Handlers map these to HTTP
@@ -41,4 +42,7 @@ var (
 	ErrUploadRegistration = errors.New(UploadRegistrationErrorName, "failed to register tenant with upload service")
 	// ErrDIDDeactivation is returned when deactivating the tenant's did:plc fails.
 	ErrDIDDeactivation = errors.New(DIDDeactivationErrorName, "failed to deactivate tenant DID")
+	// ErrExportInProgress is returned when deleting a tenant that has an open
+	// export session.
+	ErrExportInProgress = errors.New(ExportInProgressErrorName, "tenant has an export in progress")
 )

pkg/api/service/tenant/service.go

@@ -18,6 +18,7 @@ import (
 	"github.com/fil-forge/hilt/pkg/store/bucket"
 	bucketpolicystore "github.com/fil-forge/hilt/pkg/store/bucketpolicy"
 	"github.com/fil-forge/hilt/pkg/store/delegation"
+	"github.com/fil-forge/hilt/pkg/store/exportsession"
 	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"
@@ -48,6 +49,7 @@ type Service struct {
 	delegations delegation.Store
 	secrets     vault.Vault
 	wrapKeys    wrapkeystore.Store
+	exports     exportsession.Store
 	plcClient   *plc.DirectoryClient
 	upload      *upload.Client
 	revocations grant.RevocationPublisher
@@ -65,6 +67,7 @@ func New(
 	delegations delegation.Store,
 	secrets vault.Vault,
 	wrapKeys wrapkeystore.Store,
+	exports exportsession.Store,
 	plcClient *plc.DirectoryClient,
 	upload *upload.Client,
 	revocations grant.RevocationPublisher,
@@ -81,6 +84,7 @@ func New(
 		secrets:     secrets,
 		plcClient:   plcClient,
 		wrapKeys:    wrapKeys,
+		exports:     exports,
 		upload:      upload,
 		revocations: revocations,
 	}
@@ -275,9 +279,10 @@ func (s *Service) SetStatus(ctx context.Context, externalID, status string) erro
 	return nil
 }
 
-// Delete permanently deletes a tenant (which must be disabled), cascading to its
-// buckets and their policies, access keys, principals, and delegations, and
-// deactivating its did:plc. It is idempotent: a missing tenant is a no-op.
+// Delete permanently deletes a tenant (which must be disabled and have no open
+// export session), cascading to its buckets and their policies, access keys,
+// principals, delegations, and export sessions, 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.
@@ -294,6 +299,16 @@ func (s *Service) Delete(ctx context.Context, externalID string) error {
 		return ErrTenantNotDisabled
 	}
 
+	// An open export still needs the tenant's keys and buckets, and deletion
+	// destroys both, so it waits for the export to be released or aborted.
+	exporting, err := s.exports.HasOpen(ctx, rec.ID)
+	if err != nil {
+		return fmt.Errorf("checking export sessions: %w", err)
+	}
+	if exporting {
+		return ErrExportInProgress
+	}
+
 	signingVaultKey := vault.TenantKeyPath(rec.ID)
 
 	keys, err := s.accessKeys.ListByTenant(ctx, rec.ID)
@@ -399,6 +414,12 @@ func (s *Service) Delete(ctx context.Context, externalID string) error {
 		}
 	}
 
+	// Cascade: export sessions, all closed by now. They reference the tenant row
+	// (FK RESTRICT), so they go before it.
+	if err := s.exports.DeleteByTenant(ctx, rec.ID); err != nil {
+		return fmt.Errorf("deleting export sessions: %w", err)
+	}
+
 	// Delegations addressed to the tenant (the bucket -> tenant grants).
 	if err := s.delegations.DeleteByAudience(ctx, rec.ID); err != nil {
 		return fmt.Errorf("deleting tenant delegations: %w", err)

pkg/api/service/tenant/service_test.go

@@ -23,6 +23,8 @@ import (
 	bucketpolicystore "github.com/fil-forge/hilt/pkg/store/bucketpolicy"
 	bucketpolicymemory "github.com/fil-forge/hilt/pkg/store/bucketpolicy/memory"
 	delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory"
+	"github.com/fil-forge/hilt/pkg/store/exportsession"
+	exportsessionmemory "github.com/fil-forge/hilt/pkg/store/exportsession/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"
@@ -95,7 +97,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(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), plcClient, upload, &testutil.FakeSwarf{})
+		accesskeymemory.New(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), exportsessionmemory.New(), plcClient, upload, &testutil.FakeSwarf{})
 	return provisionEnv{svc: svc, providers: providers, sprueFailed: sprueFailed}
 }
 
@@ -156,7 +158,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(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), nil, nil, &testutil.FakeSwarf{})
+		accesskeymemory.New(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), exportsessionmemory.New(), nil, nil, &testutil.FakeSwarf{})
 }
 
 func TestGetAndSetStatus(t *testing.T) {
@@ -232,6 +234,7 @@ type deleteEnv struct {
 	buckets    *bucketmemory.Store
 	policies   *bucketpolicymemory.Store
 	principals *principalmemory.Store
+	exports    *exportsessionmemory.Store
 	swarf      *testutil.FakeSwarf
 	// publisher is swarf as the service sees it, recording each publish's deadline.
 	publisher *deadlinePublisher
@@ -309,10 +312,11 @@ func deleteSetup(t *testing.T, status tenant.Status) deleteEnv {
 
 	swarf := &testutil.FakeSwarf{}
 	publisher := &deadlinePublisher{FakeSwarf: swarf}
+	exports := exportsessionmemory.New()
 	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), buckets,
-		accessKeys, principals, policies, delegations, secrets, wrapkeysmemory.New(), plcClient, nil, publisher)
+		accessKeys, principals, policies, delegations, secrets, wrapkeysmemory.New(), exports, plcClient, nil, publisher)
 	return deleteEnv{svc: svc, tenants: tenants, buckets: buckets, policies: policies,
-		principals: principals, swarf: swarf, publisher: publisher, bucketID: bucketID, tenantID: tenantID, grants: grants, directory: directory,
+		principals: principals, exports: exports, swarf: swarf, publisher: publisher, bucketID: bucketID, tenantID: tenantID, grants: grants, directory: directory,
 		tombstone: tombstoneJSON.Bytes()}
 }
 
@@ -401,6 +405,25 @@ func TestDelete(t *testing.T) {
 		require.ErrorIs(t, env.svc.Delete(ctx, "tenant-1"), tenantsvc.ErrTenantNotDisabled)
 	})
 
+	t.Run("refuses while an export is open, then deletes once it closes", func(t *testing.T) {
+		env := deleteSetup(t, tenant.Disabled)
+		require.NoError(t, env.exports.Add(ctx, exportsession.Input{
+			ID: "session-1", Tenant: env.tenantID, Bucket: env.bucketID,
+			CustomerKey: "z6LSbysY2xFMRpGMhb7tFTLMpeuPRaqaWM1yECx2AtzE3KCc", ExpiresAt: time.Now().Add(time.Hour),
+		}))
+
+		require.ErrorIs(t, env.svc.Delete(ctx, "tenant-1"), tenantsvc.ErrExportInProgress)
+		_, err := env.tenants.GetByExternalID(ctx, "tenant-1")
+		require.NoError(t, err)
+		require.Equal(t, 0, env.directory.deactivations, "nothing is torn down while the export is open")
+		require.Empty(t, env.swarf.Revoked(), "no delegation is revoked while the export is open")
+
+		require.NoError(t, env.exports.Transition(ctx, "session-1", exportsession.Opened, exportsession.Aborted))
+		require.NoError(t, env.svc.Delete(ctx, "tenant-1"))
+		_, err = env.exports.Get(ctx, "session-1")
+		require.ErrorIs(t, err, store.ErrRecordNotFound, "closed sessions are deleted with the tenant")
+	})
+
 	t.Run("maps a directory failure to ErrDIDDeactivation", func(t *testing.T) {
 		env := deleteSetup(t, tenant.Disabled)
 		env.directory.logLastStatus = http.StatusInternalServerError

pkg/api/tenants.go

@@ -22,7 +22,7 @@ func tenantHTTPError(log *zap.Logger, err error) error {
 		return httpError(http.StatusBadRequest, err)
 	case errors.Is(err, tenantsvc.ErrInvalidStatus):
 		return httpError(http.StatusUnprocessableEntity, err)
-	case errors.Is(err, tenantsvc.ErrTenantNotDisabled):
+	case errors.Is(err, tenantsvc.ErrTenantNotDisabled), errors.Is(err, tenantsvc.ErrExportInProgress):
 		return httpError(http.StatusConflict, err)
 	case errors.Is(err, tenantsvc.ErrDIDRegistration),
 		errors.Is(err, tenantsvc.ErrUploadRegistration),
@@ -93,7 +93,7 @@ func NewUpdateTenantStatusHandler(logger *zap.Logger, tenants *tenantsvc.Service
 }
 
 // NewDeleteTenantHandler handles DELETE /tenants/{tenantId} — permanently delete a
-// tenant (must be disabled first). Idempotent.
+// tenant (must be disabled first, with no export in progress). Idempotent.
 func NewDeleteTenantHandler(logger *zap.Logger, tenants *tenantsvc.Service) Route {
 	log := logger.With(zap.String("handler", "DeleteTenant"))
 	return NewRoute(http.MethodDelete, "/tenants/:tenantId", func(c echo.Context) error {

pkg/api/tenants_test.go

@@ -11,6 +11,7 @@ import (
 	"net/url"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/fil-forge/hilt/internal/testutil"
 	"github.com/fil-forge/hilt/pkg/api"
@@ -22,6 +23,8 @@ import (
 	bucketmemory "github.com/fil-forge/hilt/pkg/store/bucket/memory"
 	bucketpolicymemory "github.com/fil-forge/hilt/pkg/store/bucketpolicy/memory"
 	delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory"
+	"github.com/fil-forge/hilt/pkg/store/exportsession"
+	exportsessionmemory "github.com/fil-forge/hilt/pkg/store/exportsession/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"
@@ -230,7 +233,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(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), deps.secrets, deps.wrapKeys, plcClient, upload, &testutil.FakeSwarf{})
+	svc := tenantsvc.New(zap.NewNop(), deps.tenants, deps.providers, bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), deps.secrets, deps.wrapKeys, exportsessionmemory.New(), plcClient, upload, &testutil.FakeSwarf{})
 	route := api.NewProvisionTenantHandler(zap.NewNop(), svc)
 	e := echo.New()
 	e.Add(route.Method, route.Path, route.Handler)
@@ -513,7 +516,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(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil, &testutil.FakeSwarf{})
+	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), exportsessionmemory.New(), nil, nil, &testutil.FakeSwarf{})
 	e := serve(api.NewGetTenantHandler(zap.NewNop(), svc))
 
 	t.Run("found", func(t *testing.T) {
@@ -541,7 +544,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(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil, &testutil.FakeSwarf{})
+	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), bucketpolicymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), exportsessionmemory.New(), nil, nil, &testutil.FakeSwarf{})
 	e := serve(api.NewUpdateTenantStatusHandler(zap.NewNop(), svc))
 
 	statusBody := func(s api.TenantStatus) []byte {
@@ -613,6 +616,7 @@ type deleteDeps struct {
 	principals  *principalmemory.Store
 	delegations *delegationmemory.Store
 	wrapKeys    *wrapkeymemory.Store
+	exports     *exportsessionmemory.Store
 	secrets     vault.Vault
 	directory   *plcDirectory
 	signer      secp256k1.Signer
@@ -668,6 +672,7 @@ func setupDelete(t *testing.T, status tenant.Status) (*echo.Echo, *deleteDeps) {
 		principals:  principalmemory.New(),
 		delegations: delegationmemory.New(),
 		wrapKeys:    wrapkeymemory.New(),
+		exports:     exportsessionmemory.New(),
 		secrets:     vaultmemory.New(),
 		directory:   directory,
 		signer:      signer,
@@ -689,7 +694,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.principals, bucketpolicymemory.New(), deps.delegations, deps.secrets, deps.wrapKeys, plcClient, nil, &testutil.FakeSwarf{})
+	svc := tenantsvc.New(zap.NewNop(), deps.tenants, providermemory.New(), deps.buckets, deps.accessKeys, deps.principals, bucketpolicymemory.New(), deps.delegations, deps.secrets, deps.wrapKeys, deps.exports, plcClient, nil, &testutil.FakeSwarf{})
 	route := api.NewDeleteTenantHandler(zap.NewNop(), svc)
 	return serve(route), deps
 }
@@ -765,6 +770,20 @@ func TestDeleteTenantHandler(t *testing.T) {
 		require.Equal(t, 1, deps.directory.deactivations)
 	})
 
+	t.Run("409 while an export is in progress", func(t *testing.T) {
+		e, deps := setupDelete(t, tenant.Disabled)
+		require.NoError(t, deps.exports.Add(ctx, exportsession.Input{
+			ID: "session-1", Tenant: deps.tenantID, Bucket: testutil.RandomDID(t),
+			CustomerKey: "z6LSbysY2xFMRpGMhb7tFTLMpeuPRaqaWM1yECx2AtzE3KCc", ExpiresAt: time.Now().Add(time.Hour),
+		}))
+
+		rec := doRequest(t, e, http.MethodDelete, "/tenants/tenant-1", nil)
+		require.Equal(t, http.StatusConflict, rec.Code)
+		_, err := deps.tenants.GetByExternalID(ctx, "tenant-1")
+		require.NoError(t, err)
+		require.Equal(t, 0, deps.directory.deactivations)
+	})
+
 	t.Run("already-deactivated DID still cleans up locally", func(t *testing.T) {
 		e, deps := setupDelete(t, tenant.Disabled)
 		deps.serveTombstone(t) // directory reports the DID as already tombstoned

pkg/fx/store/memory/provider.go

@@ -11,6 +11,8 @@ import (
 	membucketpolicy "github.com/fil-forge/hilt/pkg/store/bucketpolicy/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/exportsession"
+	memexportsession "github.com/fil-forge/hilt/pkg/store/exportsession/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"
@@ -32,6 +34,7 @@ var Module = fx.Module("memory-store",
 		}, fx.As(new(accesskey.Store))),
 		fx.Annotate(membucket.New, fx.As(new(bucket.Store)), fx.As(fx.Self())),
 		fx.Annotate(memdelegation.New, fx.As(new(delegation.Store))),
+		fx.Annotate(memexportsession.New, fx.As(new(exportsession.Store))),
 		// The policy store refuses a bucket the bucket store does not hold,
 		// as the Postgres backend's bucket lock does.
 		fx.Annotate(func(buckets *membucket.Store) *membucketpolicy.Store {

pkg/fx/store/postgres/provider.go

@@ -18,6 +18,8 @@ import (
 	pgbucketpolicy "github.com/fil-forge/hilt/pkg/store/bucketpolicy/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/exportsession"
+	pgexportsession "github.com/fil-forge/hilt/pkg/store/exportsession/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"
@@ -39,6 +41,7 @@ var Module = fx.Module("postgres-store",
 		NewAccessKeyStore,
 		NewBucketStore,
 		NewDelegationStore,
+		NewExportSessionStore,
 		NewPolicyStore,
 		NewPrincipalStore,
 		NewProviderStore,
@@ -126,6 +129,10 @@ func NewDelegationStore(mdb *MigratedPool) delegation.Store {
 	return pgdelegation.New(mdb.Pool)
 }
 
+func NewExportSessionStore(mdb *MigratedPool) exportsession.Store {
+	return pgexportsession.New(mdb.Pool)
+}
+
 func NewPolicyStore(mdb *MigratedPool) bucketpolicy.Store {
 	return pgbucketpolicy.New(mdb.Pool)
 }