From 1b07c58b344ac256ba112fb96b4235f171400510 Mon Sep 17 00:00:00 2001
From: eric <eric@posthog.com>
Date: Fri, 2 Oct 2026 14:58:00 -0400
Subject: [PATCH 1/2] fix(trino): retire superseded validating candidates

---
 controlplane/trino_pool_progress.go           |  40 ++++-
 controlplane/trino_pool_superseded_test.go    |  56 ++++++
 .../trino_pool_validating_race_test.go        | 168 ++++++++++++++++++
 docs/runbooks/trino-service-credentials.md    |   8 +-
 4 files changed, 270 insertions(+), 2 deletions(-)
 create mode 100644 controlplane/trino_pool_validating_race_test.go

diff --git a/controlplane/trino_pool_progress.go b/controlplane/trino_pool_progress.go
index eef7d422..582ff61d 100644
--- a/controlplane/trino_pool_progress.go
+++ b/controlplane/trino_pool_progress.go
@@ -88,7 +88,7 @@ func (o *trinoPoolOperator) progressInstance(ctx context.Context, instance confi
 			return false, err
 		}
 		// Configuration changes cannot repair an immutable candidate snapshot.
-		// VALIDATING can have an unknown admission outcome and must resolve it first.
+		// VALIDATING requires an admission read-back and guarded retirement first.
 		// Cleanup still requires the Gateway's guarded retirement claim before deletion.
 		if superseded {
 			return true, o.failCandidate(ctx, instance, trinopool.PhasePreparing,
@@ -382,6 +382,15 @@ func (o *trinoPoolOperator) validateCandidate(ctx context.Context, instance conf
 // the budgets, the certificate freshness and any open publication barrier, and
 // independently verifies the live process identity.
 func (o *trinoPoolOperator) admitCandidate(ctx context.Context, instance configstore.TrinoPoolInstance) error {
+	superseded, err := o.candidateSuperseded(instance)
+	if err != nil {
+		return err
+	}
+	if superseded {
+		if retired, err := o.retireSupersededCandidate(ctx, instance); retired || err != nil {
+			return err
+		}
+	}
 	validation, err := unmarshalValidationReceipt(instance.ValidationReceipt)
 	if err != nil {
 		return fmt.Errorf("instance %s has an unreadable validation receipt: %w", instance.InstanceID, err)
@@ -467,6 +476,35 @@ func (o *trinoPoolOperator) admitCandidate(ctx context.Context, instance configs
 		}))
 }
 
+// retireSupersededCandidate requires a Gateway claim before changing the local phase.
+// Its CAS serializes retirement with an admission whose response may be lost.
+func (o *trinoPoolOperator) retireSupersededCandidate(ctx context.Context, instance configstore.TrinoPoolInstance) (bool, error) {
+	member, err := o.gateway.GetMember(ctx, o.config.RoutingGroup, instance.InstanceID)
+	if err != nil {
+		return false, o.dropAuthority(fmt.Errorf("read superseded candidate %s: %w", instance.InstanceID, err))
+	}
+	if member.Incarnation != instance.GatewayIncarnation || member.PodUID != instance.CoordinatorPodUID || member.BootID != instance.CoordinatorBootID {
+		return false, errors.New("superseded candidate read-back identity changed")
+	}
+	switch member.Phase {
+	case "ACTIVE":
+		return false, nil
+	case "PREPARING":
+		_, err := o.gateway.RetireMember(ctx, o.config.RoutingGroup, instance.InstanceID, trinogateway.MemberStepRequest{
+			Step:               o.step(instance.InstanceID, "retire"),
+			ExpectedGeneration: member.Generation,
+		})
+		if err != nil {
+			return false, o.dropAuthority(fmt.Errorf("claim retirement of superseded candidate %s: %w", instance.InstanceID, err))
+		}
+	case "RETIRING", "RETIRED":
+		// Read-back resolves a lost retirement response or local checkpoint.
+	default:
+		return false, fmt.Errorf("superseded candidate %s is not a never-admitted member: %s", instance.InstanceID, member.Phase)
+	}
+	return true, o.failCandidate(ctx, instance, trinopool.PhaseValidating, "candidate blueprint was superseded before admission")
+}
+
 // markServing records that the Gateway considers the member eligible. Serving
 // is the Gateway's judgement, not ours: it is what the minimum-serving floor
 // counts.
diff --git a/controlplane/trino_pool_superseded_test.go b/controlplane/trino_pool_superseded_test.go
index b4b78e64..956968fc 100644
--- a/controlplane/trino_pool_superseded_test.go
+++ b/controlplane/trino_pool_superseded_test.go
@@ -370,3 +370,59 @@ func TestSupersededCandidateRetirementRequiresAuthority(t *testing.T) {
 		}
 	})
 }
+
+func TestSupersededValidatingCandidateRecoversAfterRejection(t *testing.T) {
+	for _, newRelease := range []bool{false, true} {
+		t.Run(fmt.Sprint("new release=", newRelease), func(t *testing.T) {
+			h, instance := supersededCandidateHarness(t)
+			h.tick(t, 1)
+			original := *instance
+			h.gateway.admitErr = trinogateway.ErrNotCertified
+			h.tickTolerant(2)
+			if instance.Phase != string(trinopool.PhaseValidating) {
+				t.Fatal("unchanged rejected candidate must retain its slot")
+			}
+			current := changedCandidateConfig(t, h.operator.config, newRelease)
+			h.operator.resolveConfig = func() (trinoPoolConfig, error) { return current, nil }
+			h.gateway.loseResponse = map[string]bool{"retire": true}
+			h.tickTolerant(1)
+			if h.gateway.members[instance.InstanceID].Phase != "RETIRING" || instance.Phase != string(trinopool.PhaseValidating) || h.kube.deleted[instance.InstanceID] {
+				t.Fatal("lost retirement response must retain the candidate until read-back")
+			}
+			h.gateway.admitErr = nil
+			h.tick(t, 1)
+			if instance.Phase != string(trinopool.PhaseFailedPreparing) || h.kube.deleted[instance.InstanceID] {
+				t.Fatal("superseded rejected candidate did not enter guarded cleanup")
+			}
+			if instance.ValidationReceipt != original.ValidationReceipt || instance.BlueprintSnapshot != original.BlueprintSnapshot || instance.ReleaseID != original.ReleaseID || instance.SpecDigest != original.SpecDigest {
+				t.Fatal("supersession rewrote immutable admission intent")
+			}
+			h.tick(t, 1)
+			if !h.kube.deleted[instance.InstanceID] || !trinopool.Phase(instance.Phase).OccupiesCapacity() || len(h.store.order) != 1 {
+				t.Fatal("candidate capacity released before resource absence")
+			}
+			h.kube.absent = true
+			h.tick(t, 1)
+			if instance.Phase != string(trinopool.PhaseFailureRetired) || h.gateway.members[instance.InstanceID].Phase != "RETIRED" {
+				t.Fatal("candidate cleanup did not finish")
+			}
+			h.gateway.admitErr = nil
+			h.tick(t, 6)
+			replacement := h.store.instances[h.store.order[1]]
+			if replacement.Phase != string(trinopool.PhaseServing) || replacement.ReleaseID != current.Spec.DesiredReleaseID || replacement.SpecDigest != current.Blueprint.SpecDigest(h.operator.identityFor(replacement.InstanceID)) {
+				t.Fatal("replacement did not serve the desired blueprint")
+			}
+		})
+	}
+}
+
+func TestSupersededValidatingCandidatePreservesUnreadableAdmission(t *testing.T) {
+	h, instance := supersededCandidateHarness(t)
+	h.tick(t, 1)
+	h.operator.config = changedCandidateConfig(t, h.operator.config, true)
+	h.operator.gateway = &unreadableCreatingMemberGateway{h.gateway}
+	h.tickTolerant(2)
+	if instance.Phase != string(trinopool.PhaseValidating) || h.kube.deleted[instance.InstanceID] || h.gateway.members[instance.InstanceID].Phase != "PREPARING" {
+		t.Fatal("unreadable admission outcome authorized cleanup")
+	}
+}
diff --git a/controlplane/trino_pool_validating_race_test.go b/controlplane/trino_pool_validating_race_test.go
new file mode 100644
index 00000000..51f46c50
--- /dev/null
+++ b/controlplane/trino_pool_validating_race_test.go
@@ -0,0 +1,168 @@
+//go:build kubernetes
+
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/posthog/duckgres/controlplane/configstore"
+	"github.com/posthog/duckgres/controlplane/trinogateway"
+	"github.com/posthog/duckgres/controlplane/trinopool"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func supersededValidatingFixture(t *testing.T) (*operatorHarness, *configstore.TrinoPoolInstance, trinogateway.AdmitMemberRequest) {
+	t.Helper()
+	h, instance := supersededCandidateHarness(t)
+	h.tick(t, 1)
+	current := changedCandidateConfig(t, h.operator.config, true)
+	h.operator.config = current
+	h.store.pool.DesiredReleaseID = current.Spec.DesiredReleaseID
+	h.store.pool.DesiredBlueprintDigest = current.Spec.DesiredBlueprintDigest
+	validation, err := unmarshalValidationReceipt(instance.ValidationReceipt)
+	if err != nil {
+		t.Fatal(err)
+	}
+	request := trinogateway.AdmitMemberRequest{
+		Step: h.operator.step(instance.InstanceID, "admit"), ExpectedGeneration: instance.GatewayGeneration,
+		Receipt: trinogateway.ValidationReceipt{
+			CertificateHash: validation.CertificateHash, ConfigRevision: instance.ReleaseID, AuthRevision: validation.AuthRevision,
+			PodUID: instance.CoordinatorPodUID, BootID: validation.ProcessID, NodeID: validation.NodeID,
+			CoordinatorID: validation.CoordinatorID, ReadyWorkers: validation.ReadyWorkers, Checks: validation.Checks,
+		},
+	}
+	return h, instance, request
+}
+
+func TestSupersededValidatingAdmissionWinsRetirementRace(t *testing.T) {
+	h, instance, request := supersededValidatingFixture(t)
+	race := &trinoNodeAdmissionRaceGateway{fakePoolGateway: h.gateway}
+	race.beforeRetire = func() {
+		if _, err := h.gateway.AdmitMember(context.Background(), h.operator.config.RoutingGroup, instance.InstanceID, request); err != nil {
+			t.Fatal(err)
+		}
+	}
+	h.operator.gateway = race
+	if err := h.operator.admitCandidate(context.Background(), *instance); err == nil {
+		t.Fatal("retirement did not lose its generation CAS")
+	}
+	if instance.Phase != string(trinopool.PhaseValidating) || h.gateway.members[instance.InstanceID].Phase != "ACTIVE" || len(h.kube.deleted) != 0 {
+		t.Fatal("retirement race discarded admitted work")
+	}
+	if err := h.operator.admitCandidate(context.Background(), *instance); err != nil {
+		t.Fatal(err)
+	}
+	if instance.Phase != string(trinopool.PhaseAdmitted) || len(h.kube.deleted) != 0 {
+		t.Fatal("committed admission was not adopted after retirement lost")
+	}
+}
+
+func TestSupersededValidatingRetirementWinsDelayedAdmission(t *testing.T) {
+	h, instance, request := supersededValidatingFixture(t)
+	if err := h.operator.admitCandidate(context.Background(), *instance); err != nil {
+		t.Fatal(err)
+	}
+	if instance.Phase != string(trinopool.PhaseFailedPreparing) || h.gateway.members[instance.InstanceID].Phase != "RETIRING" {
+		t.Fatal("superseded candidate did not claim retirement")
+	}
+	if _, err := h.gateway.AdmitMember(context.Background(), h.operator.config.RoutingGroup, instance.InstanceID, request); err == nil {
+		t.Fatal("delayed admission succeeded after retirement won")
+	}
+	if h.gateway.members[instance.InstanceID].Phase != "RETIRING" || len(h.kube.deleted) != 0 {
+		t.Fatal("delayed admission changed retirement or deleted resources")
+	}
+}
+
+func TestSupersededValidatingIdentityMismatchPreservesResources(t *testing.T) {
+	for _, field := range []string{"incarnation", "pod", "boot"} {
+		t.Run(field, func(t *testing.T) {
+			h, instance, _ := supersededValidatingFixture(t)
+			member := h.gateway.members[instance.InstanceID]
+			switch field {
+			case "incarnation":
+				member.Incarnation += "-other"
+			case "pod":
+				member.PodUID += "-other"
+			case "boot":
+				member.BootID += "-other"
+			}
+			if err := h.operator.admitCandidate(context.Background(), *instance); err == nil {
+				t.Fatal("changed identity was accepted")
+			}
+			if instance.Phase != string(trinopool.PhaseValidating) || member.Phase != "PREPARING" || len(h.kube.deleted) != 0 {
+				t.Fatal("changed identity authorized lifecycle progress")
+			}
+		})
+	}
+}
+
+type supersededRetirementFencedGateway struct {
+	*fakePoolGateway
+}
+
+func (g *supersededRetirementFencedGateway) RetireMember(context.Context, string, string, trinogateway.MemberStepRequest) (trinogateway.Member, error) {
+	return trinogateway.Member{}, trinogateway.ErrStaleEpoch
+}
+
+func TestSupersededValidatingRetirementAuthorityLost(t *testing.T) {
+	h, instance, _ := supersededValidatingFixture(t)
+	h.operator.gateway = &supersededRetirementFencedGateway{h.gateway}
+	err := h.operator.admitCandidate(context.Background(), *instance)
+	if !errors.Is(err, trinogateway.ErrStaleEpoch) || !h.operator.fenced {
+		t.Fatalf("lost retirement authority did not fence the operator: %v", err)
+	}
+	if instance.Phase != string(trinopool.PhaseValidating) || h.gateway.members[instance.InstanceID].Phase != "PREPARING" || len(h.kube.deleted) != 0 {
+		t.Fatal("lost authority authorized cleanup")
+	}
+}
+
+func TestSupersededValidatingRetirementRecoversFailedCheckpoint(t *testing.T) {
+	h, instance, _ := supersededValidatingFixture(t)
+	h.store.failAdvance = true
+	if err := h.operator.admitCandidate(context.Background(), *instance); !errors.Is(err, configstore.ErrTrinoPoolConflict) {
+		t.Fatalf("expected failed phase checkpoint: %v", err)
+	}
+	if instance.Phase != string(trinopool.PhaseValidating) || h.gateway.members[instance.InstanceID].Phase != "RETIRING" || len(h.kube.deleted) != 0 {
+		t.Fatal("failed checkpoint lost the guarded retirement state")
+	}
+	h.store.failAdvance = false
+	h.operator.fenced = false
+	previousEpoch := h.store.epoch
+	h.tick(t, 1)
+	if h.operator.lease.Epoch <= previousEpoch {
+		t.Fatal("new leadership term did not reacquire authority")
+	}
+	if instance.Phase != string(trinopool.PhaseFailedPreparing) || len(h.kube.deleted) != 0 {
+		t.Fatal("retirement read-back did not recover the failed checkpoint")
+	}
+}
+
+func TestSupersededValidatingRetirementRecoversAfterConfigRollback(t *testing.T) {
+	h, instance, _ := supersededValidatingFixture(t)
+	h.operator.nodeGuard = newTrinoPoolNodeGuard(fake.NewClientset())
+	h.gateway.loseResponse = map[string]bool{"retire": true}
+	if err := h.operator.admitCandidate(context.Background(), *instance); err == nil {
+		t.Fatal("retirement response was not lost")
+	}
+	if instance.Phase != string(trinopool.PhaseValidating) || h.gateway.members[instance.InstanceID].Phase != "RETIRING" || len(h.kube.deleted) != 0 {
+		t.Fatal("lost retirement response did not retain the candidate for read-back")
+	}
+	blueprint, err := trinopool.ParseBlueprint([]byte(instance.BlueprintSnapshot))
+	if err != nil {
+		t.Fatal(err)
+	}
+	h.operator.config.Blueprint = blueprint
+	h.operator.config.Spec.DesiredReleaseID = instance.ReleaseID
+	h.operator.config.Spec.DesiredBlueprintDigest = blueprint.Digest()
+	if err := h.store.UpsertTrinoPoolSpec(context.Background(), h.operator.lease, h.operator.config.Spec); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.operator.admitCandidate(context.Background(), *instance); err != nil {
+		t.Fatal(err)
+	}
+	if instance.Phase != string(trinopool.PhaseFailedPreparing) || h.gateway.members[instance.InstanceID].Phase != "RETIRING" || len(h.kube.deleted) != 0 {
+		t.Fatal("configuration rollback bypassed committed retirement read-back")
+	}
+}
diff --git a/docs/runbooks/trino-service-credentials.md b/docs/runbooks/trino-service-credentials.md
index e8074bd3..69d3edfa 100644
--- a/docs/runbooks/trino-service-credentials.md
+++ b/docs/runbooks/trino-service-credentials.md
@@ -126,4 +126,10 @@ This callback check proves reachability and pool-token acceptance, not tenant-gr
 Deploy the Duckgres readiness endpoint and controller checks before the Trino reporting implementation.
 Then publish the new Trino image through the normal blueprint rollout and verify the disposable service-credential fixture lane with `TRINO_SERVICE_CREDENTIALS_ENABLED=true`.
 Existing `VALIDATING` candidates retain their durable receipt; this change does not rewrite or replay it with a different intent.
-Use the existing guarded recovery procedure separately if such a candidate blocks replacement.
+When a candidate's immutable blueprint no longer matches the desired release or configuration, the controller reads the candidate's Gateway state before retrying admission.
+An already-active member resolves its original admission and follows the normal serving rollout.
+A never-admitted `PREPARING` member must obtain the Gateway's generation-guarded retirement claim before entering `FAILED_PREPARING` cleanup.
+A concurrent admission, unavailable Gateway, or changed identity prevents cleanup; a lost retirement response resumes through read-back.
+The candidate retains capacity until Kubernetes confirms resource absence and Gateway retirement completes.
+To verify recovery, observe the obsolete candidate reach `FAILURE_RETIRED`, then verify that its replacement uses the desired blueprint and reaches `SERVING`.
+Do not rewrite validation receipts or delete candidate pods to bypass this lifecycle.

From 6ceb95a5965339188289830e6d4ab5396b9c1778 Mon Sep 17 00:00:00 2001
From: eric <eric@posthog.com>
Date: Fri, 2 Oct 2026 15:56:03 -0400
Subject: [PATCH 2/2] fix(trino): gate tenant admission on catalog and storage
 readiness

---
 controlplane/provisioner/trino_hoglake.go     |  39 +++-
 .../provisioner/trino_hoglake_storage.go      | 114 +++++++++
 .../provisioner/trino_hoglake_storage_test.go | 221 ++++++++++++++++++
 controlplane/provisioner/trino_provisioner.go |   9 +-
 .../provisioner/trino_provisioner_test.go     |   1 +
 controlplane/trino_inputs.go                  |  15 ++
 controlplane/trino_pool_catalog.go            |  20 ++
 .../trino_pool_catalog_admission_test.go      | 152 ++++++++++++
 ...rino_pool_catalog_hoglake_postgres_test.go |  54 +++++
 controlplane/trino_pool_operator.go           |   3 +
 controlplane/trino_pool_operator_test.go      |   3 +
 controlplane/trino_pool_publication.go        |  42 +++-
 controlplane/trino_pool_wiring.go             |   1 +
 docs/runbooks/trino-hoglake-provisioning.md   |  26 ++-
 14 files changed, 672 insertions(+), 28 deletions(-)
 create mode 100644 controlplane/provisioner/trino_hoglake_storage.go
 create mode 100644 controlplane/provisioner/trino_hoglake_storage_test.go
 create mode 100644 controlplane/trino_pool_catalog_admission_test.go

diff --git a/controlplane/provisioner/trino_hoglake.go b/controlplane/provisioner/trino_hoglake.go
index 75db3cd0..386fab21 100644
--- a/controlplane/provisioner/trino_hoglake.go
+++ b/controlplane/provisioner/trino_hoglake.go
@@ -29,6 +29,8 @@ type TrinoManagedHoglakeConfig struct {
 	Namespace string
 }
 
+var ErrHoglakeStorageNotReady = errors.New("hoglake tenant storage is not ready")
+
 var hoglakeIdentifier = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
 
 // The trusted, configured service origin is internal infrastructure traffic.
@@ -44,14 +46,8 @@ func (c TrinoManagedHoglakeConfig) Validate() error {
 	if err != nil || u.Hostname() == "" || (u.Scheme != "http" && u.Scheme != "https") || u.User != nil || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" || u.RawPath != "" || (u.Path != "" && u.Path != "/") {
 		return errors.New("managed Hoglake URI must be an HTTP(S) origin without credentials")
 	}
-	p, err := url.Parse(c.DataPath)
-	if err != nil || p.Scheme != "s3" || p.Hostname() == "" || p.Host != p.Hostname() || p.User != nil || p.RawQuery != "" || p.ForceQuery || p.Fragment != "" || p.RawPath != "" || strings.ContainsAny(c.DataPath, "%\t\r\n ") || !strings.HasSuffix(p.Path, "/") || p.Path == "/" || p.Path == "" || strings.Contains(p.Path, "//") {
-		return errors.New("managed Hoglake data path must be a dedicated s3://bucket/prefix/ URI")
-	}
-	for _, part := range strings.Split(strings.Trim(p.Path, "/"), "/") {
-		if part == "" || part == "." || part == ".." {
-			return errors.New("managed Hoglake data path has an invalid segment")
-		}
+	if _, err := parseHoglakeDataPath(c.DataPath); err != nil {
+		return err
 	}
 	if !hoglakeIdentifier.MatchString(c.Namespace) {
 		return errors.New("managed Hoglake namespace must contain only letters, digits, underscores or hyphens")
@@ -59,6 +55,19 @@ func (c TrinoManagedHoglakeConfig) Validate() error {
 	return nil
 }
 
+func parseHoglakeDataPath(dataPath string) (*url.URL, error) {
+	p, err := url.Parse(dataPath)
+	if err != nil || p.Scheme != "s3" || p.Hostname() == "" || p.Host != p.Hostname() || p.User != nil || p.RawQuery != "" || p.ForceQuery || p.Fragment != "" || p.RawPath != "" || strings.ContainsAny(dataPath, "%\t\r\n ") || !strings.HasSuffix(p.Path, "/") || p.Path == "/" || p.Path == "" || strings.Contains(p.Path, "//") {
+		return nil, errors.New("managed Hoglake data path must be a dedicated s3://bucket/prefix/ URI")
+	}
+	for _, part := range strings.Split(strings.Trim(p.Path, "/"), "/") {
+		if part == "" || part == "." || part == ".." {
+			return nil, errors.New("managed Hoglake data path has an invalid segment")
+		}
+	}
+	return p, nil
+}
+
 func (c TrinoManagedHoglakeConfig) catalogPath(orgID string) (string, error) {
 	if err := c.Validate(); err != nil {
 		return "", err
@@ -185,14 +194,14 @@ func isManagedHoglake(org configstore.TrinoEnabledOrg) bool {
 
 // ManagedHoglakeConfigured reports whether this provisioner can provision a
 // managed Hoglake tenant at all: the service configuration AND the storage
-// resolver that supplies the tenant's IAM role and region.
+// resolver that supplies the tenant's IAM role and region, plus the initial storage check.
 //
 // It is exported for the startup wiring's own test. A cell that silently lost
 // either input builds and reconciles perfectly until the first Hoglake tenant
 // is provisioned, and then holds that warehouse pending with an error about
 // configuration nobody changed. Test these dependencies at the pool wiring boundary.
 func (p *TrinoProvisioner) ManagedHoglakeConfigured() bool {
-	return p.managedHoglake != nil && p.hoglakeDucklings != nil
+	return p.managedHoglake != nil && p.hoglakeDucklings != nil && p.hoglakeStorageCheck != nil
 }
 
 func (p *TrinoProvisioner) managedHoglakeProperties(orgID string, d *DucklingStatus) (map[string]string, error) {
@@ -263,6 +272,16 @@ func (p *TrinoProvisioner) reconcileHoglakeCatalog(ctx context.Context, client T
 		return err
 	}
 	if !exists {
+		if p.hoglakeStorageCheck == nil {
+			return errors.New("hoglake storage readiness checker is unavailable")
+		}
+		dataPath, pathErr := p.managedHoglake.catalogPath(warehouse.DucklingName)
+		if pathErr != nil {
+			return pathErr
+		}
+		if err = p.hoglakeStorageCheck(ctx, status.IAMRoleARN, props["s3.region"], dataPath); err != nil {
+			return fmt.Errorf("%w: %s", ErrHoglakeStorageNotReady, err)
+		}
 		if err = client.CreateCatalog(ctx, name, props); err != nil {
 			return err
 		}
diff --git a/controlplane/provisioner/trino_hoglake_storage.go b/controlplane/provisioner/trino_hoglake_storage.go
new file mode 100644
index 00000000..961ad8cd
--- /dev/null
+++ b/controlplane/provisioner/trino_hoglake_storage.go
@@ -0,0 +1,114 @@
+//go:build kubernetes
+
+package provisioner
+
+import (
+	"context"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/google/uuid"
+)
+
+const hoglakeStorageProbeBody = "duckgres storage readiness\n"
+
+// TrinoHoglakeStorageCheck verifies a tenant role can use its exact catalog prefix.
+type TrinoHoglakeStorageCheck func(ctx context.Context, roleARN, region, dataPath string) error
+
+type hoglakeStorageClient interface {
+	PutObject(context.Context, *s3.PutObjectInput, ...func(*s3.Options)) (*s3.PutObjectOutput, error)
+	GetObject(context.Context, *s3.GetObjectInput, ...func(*s3.Options)) (*s3.GetObjectOutput, error)
+	DeleteObject(context.Context, *s3.DeleteObjectInput, ...func(*s3.Options)) (*s3.DeleteObjectOutput, error)
+}
+
+// HoglakeStorageProbe gates initial catalog publication on usable tenant storage.
+// Failed attempts reuse their object key; concurrent control planes have separate keys.
+type HoglakeStorageProbe struct {
+	assume     AssumeRoleFunc
+	instanceID string
+	newClient  func(context.Context, string, string, string, string) (hoglakeStorageClient, error)
+}
+
+func NewHoglakeStorageProbe(assume AssumeRoleFunc) *HoglakeStorageProbe {
+	return &HoglakeStorageProbe{assume: assume, instanceID: uuid.NewString(), newClient: newHoglakeStorageClient}
+}
+
+func newHoglakeStorageClient(ctx context.Context, region, accessKey, secretKey, token string) (hoglakeStorageClient, error) {
+	cfg, err := awsconfig.LoadDefaultConfig(ctx,
+		awsconfig.WithRegion(region),
+		awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKey, secretKey, token)),
+		awsconfig.WithRetryMaxAttempts(1))
+	if err != nil {
+		return nil, err
+	}
+	return s3.NewFromConfig(cfg), nil
+}
+
+func (p *HoglakeStorageProbe) Check(ctx context.Context, roleARN, region, dataPath string) error {
+	if p.assume == nil || roleARN == "" || region == "" {
+		return errors.New("hoglake storage readiness requires tenant credentials and region")
+	}
+	location, err := parseHoglakeDataPath(dataPath)
+	if err != nil {
+		return errors.New("invalid Hoglake storage readiness path")
+	}
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	accessKey, secretKey, token, err := p.assume(ctx, roleARN)
+	if err != nil || accessKey == "" || secretKey == "" || token == "" {
+		return errors.New("waiting for Hoglake tenant storage credentials")
+	}
+	client, err := p.newClient(ctx, region, accessKey, secretKey, token)
+	if err != nil {
+		return errors.New("cannot initialize Hoglake storage readiness client")
+	}
+	digest := sha256.Sum256([]byte(roleARN + "\n" + region + "\n" + dataPath))
+	key := strings.TrimPrefix(location.Path, "/") + ".duckgres-readiness/" + p.instanceID + "-" + fmt.Sprintf("%x", digest)
+	bucket := location.Host
+	_, writeErr := client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), Body: strings.NewReader(hoglakeStorageProbeBody)})
+	var readErr error
+	if writeErr == nil {
+		readErr = readHoglakeStorageProbe(ctx, client, bucket, key)
+	}
+	// A timed-out PUT may have succeeded; cleanup must outlive the attempt's context.
+	cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
+	defer cleanupCancel()
+	_, deleteErr := client.DeleteObject(cleanupCtx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)})
+	switch {
+	case writeErr != nil:
+		return errors.New("waiting for Hoglake tenant storage write access")
+	case readErr != nil:
+		return errors.New("waiting for Hoglake tenant storage read access")
+	case deleteErr != nil:
+		return errors.New("waiting for Hoglake tenant storage delete access")
+	default:
+		return nil
+	}
+}
+
+func readHoglakeStorageProbe(ctx context.Context, client hoglakeStorageClient, bucket, key string) error {
+	result, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)})
+	if err != nil {
+		return err
+	}
+	if result == nil || result.Body == nil {
+		return errors.New("missing readiness object")
+	}
+	defer func() { _ = result.Body.Close() }()
+	body, err := io.ReadAll(io.LimitReader(result.Body, int64(len(hoglakeStorageProbeBody)+1)))
+	if err != nil {
+		return err
+	}
+	if string(body) != hoglakeStorageProbeBody {
+		return errors.New("readiness object content mismatch")
+	}
+	return nil
+}
diff --git a/controlplane/provisioner/trino_hoglake_storage_test.go b/controlplane/provisioner/trino_hoglake_storage_test.go
new file mode 100644
index 00000000..1d714ae4
--- /dev/null
+++ b/controlplane/provisioner/trino_hoglake_storage_test.go
@@ -0,0 +1,221 @@
+//go:build kubernetes
+
+package provisioner
+
+import (
+	"context"
+	"errors"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/posthog/duckgres/controlplane/configstore"
+	"io"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestTrinoHoglakeRequiresStorageReadiness(t *testing.T) {
+	h, client, _, _ := newHoglakeStateHarness(t)
+	h.provisioner.hoglakeStorageCheck = nil
+	err := h.provisioner.reconcileHoglakeCatalog(context.Background(), client, "org_tenant_a", "tenant-a", h.ducklings["tenant-a"], false, nil)
+	if err == nil || len(h.catalog.created) != 0 {
+		t.Fatalf("catalog published without a tenant storage readiness check: err=%v created=%v", err, h.catalog.created)
+	}
+}
+
+type fakeHoglakeStorage struct {
+	t       *testing.T
+	fail    string
+	calls   []string
+	keys    []string
+	corrupt bool
+	cancel  context.CancelFunc
+}
+
+func (f *fakeHoglakeStorage) PutObject(ctx context.Context, input *s3.PutObjectInput, _ ...func(*s3.Options)) (*s3.PutObjectOutput, error) {
+	f.calls = append(f.calls, "put")
+	f.keys = append(f.keys, aws.ToString(input.Key))
+	if aws.ToString(input.Bucket) != "example-bucket" || !strings.HasPrefix(aws.ToString(input.Key), "trino/warehouse-a/.duckgres-readiness/") || input.Tagging != nil {
+		f.t.Fatalf("unsafe readiness upload: %v", input)
+	}
+	deadline, ok := ctx.Deadline()
+	if !ok || time.Until(deadline) > 5*time.Second {
+		f.t.Fatal("unbounded storage attempt")
+	}
+	if f.cancel != nil {
+		f.cancel()
+	}
+	if f.fail == "put" {
+		return nil, errors.New("sensitive storage detail")
+	}
+	return &s3.PutObjectOutput{}, nil
+}
+
+func (f *fakeHoglakeStorage) GetObject(_ context.Context, input *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
+	f.calls = append(f.calls, "get")
+	if aws.ToString(input.Key) != f.keys[len(f.keys)-1] {
+		f.t.Fatal("read used different key")
+	}
+	if f.fail == "get" {
+		return nil, errors.New("sensitive storage detail")
+	}
+	body := hoglakeStorageProbeBody
+	if f.corrupt {
+		body += strings.Repeat("x", 1024)
+	}
+	return &s3.GetObjectOutput{Body: io.NopCloser(strings.NewReader(body))}, nil
+}
+
+func (f *fakeHoglakeStorage) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput, _ ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) {
+	f.calls = append(f.calls, "delete")
+	if aws.ToString(input.Key) != f.keys[len(f.keys)-1] {
+		f.t.Fatal("cleanup used different key")
+	}
+	deadline, ok := ctx.Deadline()
+	if ctx.Err() != nil || !ok || time.Until(deadline) > 3*time.Second {
+		f.t.Fatal("cleanup must be independently bounded")
+	}
+	if f.fail == "delete" {
+		return nil, errors.New("sensitive storage detail")
+	}
+	return &s3.DeleteObjectOutput{}, nil
+}
+
+func testHoglakeStorageProbe(t *testing.T, client *fakeHoglakeStorage) *HoglakeStorageProbe {
+	t.Helper()
+	probe := NewHoglakeStorageProbe(func(_ context.Context, role string) (string, string, string, error) {
+		if role != "tenant-role" {
+			t.Fatalf("wrong role: %s", role)
+		}
+		return "fixture-access", "fixture-secret", "fixture-session", nil
+	})
+	probe.newClient = func(_ context.Context, region, access, secret, token string) (hoglakeStorageClient, error) {
+		if region != "us-east-1" || access != "fixture-access" || secret != "fixture-secret" || token != "fixture-session" {
+			t.Fatal("tenant credentials were not passed to client")
+		}
+		return client, nil
+	}
+	return probe
+}
+
+func TestTrinoHoglakeStorageProbePermissions(t *testing.T) {
+	for _, failure := range []string{"", "put", "get", "delete", "content"} {
+		t.Run(failure, func(t *testing.T) {
+			client := &fakeHoglakeStorage{t: t, fail: failure, corrupt: failure == "content"}
+			probe := testHoglakeStorageProbe(t, client)
+			err := probe.Check(context.Background(), "tenant-role", "us-east-1", "s3://example-bucket/trino/warehouse-a/")
+			if (err != nil) != (failure != "") {
+				t.Fatalf("failure=%s err=%v", failure, err)
+			}
+			if err != nil && strings.Contains(err.Error(), "sensitive") {
+				t.Fatal("AWS details escaped readiness check")
+			}
+			want := []string{"put", "get", "delete"}
+			if failure == "put" {
+				want = []string{"put", "delete"}
+			}
+			if !slices.Equal(client.calls, want) {
+				t.Fatalf("calls=%v want=%v", client.calls, want)
+			}
+			client.fail, client.corrupt = "", false
+			if err := probe.Check(context.Background(), "tenant-role", "us-east-1", "s3://example-bucket/trino/warehouse-a/"); err != nil {
+				t.Fatal(err)
+			}
+			if client.keys[0] != client.keys[1] {
+				t.Fatal("retry created another readiness object")
+			}
+		})
+	}
+}
+
+func TestTrinoHoglakeStorageProbeCleanupAfterCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	client := &fakeHoglakeStorage{t: t, fail: "put", cancel: cancel}
+	probe := testHoglakeStorageProbe(t, client)
+	if err := probe.Check(ctx, "tenant-role", "us-east-1", "s3://example-bucket/trino/warehouse-a/"); err == nil {
+		t.Fatal("failed upload accepted")
+	}
+	if !slices.Equal(client.calls, []string{"put", "delete"}) {
+		t.Fatalf("cleanup missing: %v", client.calls)
+	}
+}
+
+func TestTrinoHoglakeStorageProbeFailsClosed(t *testing.T) {
+	for _, assume := range []AssumeRoleFunc{nil, func(context.Context, string) (string, string, string, error) { return "", "", "", nil }} {
+		probe := NewHoglakeStorageProbe(assume)
+		probe.newClient = func(context.Context, string, string, string, string) (hoglakeStorageClient, error) {
+			t.Fatal("constructed client without tenant credentials")
+			return nil, nil
+		}
+		if err := probe.Check(context.Background(), "tenant-role", "us-east-1", "s3://example-bucket/trino/warehouse-a/"); err == nil {
+			t.Fatal("absent credentials accepted")
+		}
+	}
+}
+
+func TestTrinoHoglakeStorageReadinessGatesPublication(t *testing.T) {
+	h, client, _, _ := newHoglakeStateHarness(t)
+	attempts := 0
+	denied := true
+	h.provisioner.hoglakeStorageCheck = func(_ context.Context, role, region, path string) error {
+		attempts++
+		if role != h.ducklings["tenant-a"].IAMRoleARN || region != "us-east-1" || path != "s3://example-bucket/trino/warehouse-a/" {
+			t.Fatalf("incorrect storage scope: %s %s %s", role, region, path)
+		}
+		if denied {
+			return errors.New("waiting for storage")
+		}
+		return nil
+	}
+	reconcile := func(exists bool) error {
+		return h.provisioner.reconcileHoglakeCatalog(context.Background(), client, "org_tenant_a", "tenant-a", h.ducklings["tenant-a"], exists, map[string]string{"org_tenant_a": "hoglake"})
+	}
+	if err := reconcile(false); err == nil || len(h.catalog.created) != 0 {
+		t.Fatal("denied storage published catalog")
+	}
+	denied = false
+	if err := reconcile(false); err != nil {
+		t.Fatal(err)
+	}
+	if len(h.catalog.created) != 1 || attempts != 2 {
+		t.Fatalf("publication after recovery: attempts=%d created=%v", attempts, h.catalog.created)
+	}
+	denied = true
+	if err := reconcile(true); err != nil || attempts != 2 {
+		t.Fatalf("existing catalog re-probed: attempts=%d err=%v", attempts, err)
+	}
+}
+
+func TestTrinoHoglakeStorageDenialRemainsProvisioning(t *testing.T) {
+	h, client, _, _ := newHoglakeStateHarness(t)
+	h.store.orgs[0].CellID = testCellID
+	h.store.orgs[0].Tier = "free"
+	h.store.orgs[0].RootPasswordHash = "$2a$10$hash"
+	h.provisioner.catalog = client
+	h.provisioner.hoglakeDucklings = h.provisioner.ducklings
+	h.ducklings["tenant-a"].ReadyCondition = true
+	denied := true
+	h.provisioner.hoglakeStorageCheck = func(context.Context, string, string, string) error {
+		if denied {
+			return errors.New("waiting for storage access")
+		}
+		return nil
+	}
+	if err := h.provisioner.Reconcile(context.Background()); err != nil {
+		t.Fatalf("storage propagation failed global reconcile: %v", err)
+	}
+	state, ok := h.store.lastState("tenant-a")
+	if !ok || state.State != configstore.ManagedWarehouseStateProvisioning || len(h.catalog.created) != 0 {
+		t.Fatalf("storage-denied tenant state: %+v", state)
+	}
+	denied = false
+	if err := h.provisioner.Reconcile(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	state, ok = h.store.lastState("tenant-a")
+	if !ok || state.State != configstore.ManagedWarehouseStateReady || len(h.catalog.created) != 1 {
+		t.Fatalf("storage-recovered tenant state: %+v", state)
+	}
+}
diff --git a/controlplane/provisioner/trino_provisioner.go b/controlplane/provisioner/trino_provisioner.go
index fc7861f0..80b0f348 100644
--- a/controlplane/provisioner/trino_provisioner.go
+++ b/controlplane/provisioner/trino_provisioner.go
@@ -345,7 +345,8 @@ type TrinoProvisionerOpts struct {
 	// catalogs. Defaults to false. Trino nodes must configure a cache manager.
 	FilesystemCacheEnabled bool
 
-	ManagedHoglake *TrinoManagedHoglakeConfig
+	ManagedHoglake      *TrinoManagedHoglakeConfig
+	HoglakeStorageCheck TrinoHoglakeStorageCheck
 }
 
 // TrinoBootstrapSentinelStore is the narrow configstore surface the
@@ -434,6 +435,7 @@ type TrinoProvisioner struct {
 	s3MaxConnections       int
 	filesystemCacheEnabled bool
 	managedHoglake         *TrinoManagedHoglakeConfig
+	hoglakeStorageCheck    TrinoHoglakeStorageCheck
 
 	// adminPasswordHash is cached on each Reconcile from the
 	// trino-auth K8s Secret and prepended to password.db on projection.
@@ -550,6 +552,7 @@ func NewTrinoProvisioner(opts TrinoProvisionerOpts) (*TrinoProvisioner, error) {
 		s3MaxConnections:       maxConns,
 		filesystemCacheEnabled: opts.FilesystemCacheEnabled,
 		managedHoglake:         opts.ManagedHoglake,
+		hoglakeStorageCheck:    opts.HoglakeStorageCheck,
 	}
 	return result, nil
 }
@@ -1611,7 +1614,9 @@ func (p *TrinoProvisioner) reconcileBackendCatalogs(
 			if err == nil {
 				err = p.reconcileHoglakeCatalog(ctx, catalog, name, o.OrgID, tenants.statuses[o.OrgID], existingSet[name], connectors)
 			}
-			if err != nil {
+			if errors.Is(err, ErrHoglakeStorageNotReady) {
+				outcomes[o.OrgID] = catalogOutcome{Pending: true, PendingReason: err.Error()}
+			} else if err != nil {
 				outcomes[o.OrgID] = catalogOutcome{Err: err}
 				errs = append(errs, err)
 			} else {
diff --git a/controlplane/provisioner/trino_provisioner_test.go b/controlplane/provisioner/trino_provisioner_test.go
index c80d66d1..e2f1c4e0 100644
--- a/controlplane/provisioner/trino_provisioner_test.go
+++ b/controlplane/provisioner/trino_provisioner_test.go
@@ -776,6 +776,7 @@ func newTestTrinoProvisioner(t *testing.T, orgs []configstore.TrinoEnabledOrg, w
 	if err != nil {
 		t.Fatalf("NewTrinoProvisioner: %v", err)
 	}
+	p.hoglakeStorageCheck = func(context.Context, string, string, string) error { return nil }
 	h.provisioner = p
 	return h
 }
diff --git a/controlplane/trino_inputs.go b/controlplane/trino_inputs.go
index 0bd47609..4303aa0f 100644
--- a/controlplane/trino_inputs.go
+++ b/controlplane/trino_inputs.go
@@ -160,6 +160,20 @@ func buildTrinoCellWiring(store trinoWiringStore, kc kubernetes.Interface, duckl
 	if len(storageResolvers) > 0 {
 		storageResolver = storageResolvers[0]
 	}
+	var storageCheck provisioner.TrinoHoglakeStorageCheck
+	if managedHoglake != nil {
+		broker, brokerErr := NewSTSBroker(context.Background(), strings.TrimSpace(os.Getenv(envTrinoAWSRegion)))
+		if brokerErr != nil {
+			return nil, fmt.Errorf("construct Hoglake storage credential broker: %w", brokerErr)
+		}
+		storageCheck = provisioner.NewHoglakeStorageProbe(func(ctx context.Context, roleARN string) (string, string, string, error) {
+			creds, assumeErr := broker.AssumeRole(ctx, roleARN)
+			if assumeErr != nil {
+				return "", "", "", assumeErr
+			}
+			return creds.AccessKeyID, creds.SecretAccessKey, creds.SessionToken, nil
+		}).Check
+	}
 	bundleStore := &opa.BundleStore{}
 
 	trinoProv, err := provisioner.NewTrinoProvisioner(provisioner.TrinoProvisionerOpts{
@@ -179,6 +193,7 @@ func buildTrinoCellWiring(store trinoWiringStore, kc kubernetes.Interface, duckl
 		S3MaxConnections:       envInt(envTrinoS3MaxConnections),
 		FilesystemCacheEnabled: filesystemCacheEnabled,
 		ManagedHoglake:         managedHoglake,
+		HoglakeStorageCheck:    storageCheck,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("construct Trino provisioner: %w", err)
diff --git a/controlplane/trino_pool_catalog.go b/controlplane/trino_pool_catalog.go
index c73b427d..698065b3 100644
--- a/controlplane/trino_pool_catalog.go
+++ b/controlplane/trino_pool_catalog.go
@@ -157,6 +157,26 @@ func (w *trinoPoolCatalogWriter) PublishedRevision(ctx context.Context) (int64,
 	return state.Revision, nil
 }
 
+// CatalogPublicationRevision reads inclusion and revision in one database snapshot,
+// so admission cannot certify a revision that predates this tenant's catalog.
+func (w *trinoPoolCatalogWriter) CatalogPublicationRevision(ctx context.Context, catalog string) (int64, bool, error) {
+	ctx, cancel := context.WithTimeout(ctx, trinocatalog.DefaultTimeout)
+	defer cancel()
+	var revision int64
+	err := w.db.QueryRowContext(ctx, `
+		SELECT writer.revision
+		FROM trino_catalog_writer_state AS writer
+		JOIN trino_catalogs AS catalog ON catalog.cell_id = writer.cell_id
+		WHERE writer.cell_id = $1 AND catalog.catalog_name = $2`, w.cellID, catalog).Scan(&revision)
+	if errors.Is(err, sql.ErrNoRows) {
+		return 0, false, nil
+	}
+	if err != nil {
+		return 0, false, fmt.Errorf("read tenant catalog publication: %w", err)
+	}
+	return revision, true, nil
+}
+
 // checkpoint records a revision on the pool row under the CURRENT authority.
 func (w *trinoPoolCatalogWriter) checkpoint(ctx context.Context, revision int64) error {
 	lease, held := w.authority()
diff --git a/controlplane/trino_pool_catalog_admission_test.go b/controlplane/trino_pool_catalog_admission_test.go
new file mode 100644
index 00000000..a47cab05
--- /dev/null
+++ b/controlplane/trino_pool_catalog_admission_test.go
@@ -0,0 +1,152 @@
+//go:build kubernetes
+
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/posthog/duckgres/controlplane/configstore"
+)
+
+func TestTrinoTenantAdmissionWaitsForItsPublishedCatalog(t *testing.T) {
+	h := newOperatorHarness(t)
+	h.operator.config.Pool.TenantAdmission = true
+	h.servingPool(t)
+	org := poolOrg("analyst")
+	h.operator.tenants = &fakeTenantStore{orgs: []configstore.TrinoEnabledOrg{org}}
+	published, applied, included := int64(1), int64(1), false
+	h.operator.catalogWatermark = func(context.Context) (int64, error) { return published, nil }
+	h.operator.catalogPublicationRevision = func(_ context.Context, catalog string) (int64, bool, error) {
+		if catalog != configstore.TrinoCatalogName(org.DatabaseName) {
+			t.Fatalf("catalog = %q, want the tenant's catalog", catalog)
+		}
+		if !included {
+			return 0, false, nil
+		}
+		return published, true, nil
+	}
+	h.operator.acknowledgement = func(_ context.Context, _ string, _ trinoPoolProjectionRevisions, required int64) (trinoPoolAcknowledgement, error) {
+		if required != 2 {
+			t.Fatalf("admission requires revision %d, want the revision containing the tenant (2)", required)
+		}
+		if applied < required {
+			return trinoPoolAcknowledgement{}, fmt.Errorf("catalog revision %d is not applied", required)
+		}
+		return trinoPoolAcknowledgement{ProcessID: "process-1", AppliedRevision: applied, ProjectionCurrent: true}, nil
+	}
+
+	h.tick(t, 12)
+	if row := h.publications.rows[org.OrgID]; row == nil || row.PublicationID != "" || h.gateway.admitted[org.OrgID] != "" {
+		t.Fatalf("tenant without a published catalog opened admission: %+v", row)
+	}
+	included, published = true, 2
+	h.tick(t, 12)
+	if h.gateway.admitted[org.OrgID] != "" {
+		t.Fatal("tenant admitted before members applied its catalog")
+	}
+	applied = 2
+	h.admitAll(t, 1)
+	admitted := h.gateway.admitted[org.OrgID]
+	opens := countGatewayCalls(h.gateway.calls, "open:")
+	published = 3
+	h.tick(t, 12)
+	if h.gateway.admitted[org.OrgID] != admitted || countGatewayCalls(h.gateway.calls, "open:") != opens {
+		t.Fatal("an unrelated catalog publication re-admitted the tenant")
+	}
+}
+
+func TestTrinoTenantAdmissionFailsClosedOnCatalogRead(t *testing.T) {
+	h := newOperatorHarness(t)
+	h.operator.config.Pool.TenantAdmission = true
+	h.servingPool(t)
+	h.operator.tenants = &fakeTenantStore{orgs: []configstore.TrinoEnabledOrg{poolOrg("analyst")}}
+	h.operator.catalogPublicationRevision = func(context.Context, string) (int64, bool, error) {
+		return 0, false, errors.New("catalog store unavailable")
+	}
+	h.tickTolerant(12)
+	if h.gateway.admitted["org-a"] != "" || countGatewayCalls(h.gateway.calls, "open:") != 0 {
+		t.Fatal("unreadable tenant catalog allowed admission")
+	}
+	h.operator.catalogPublicationRevision = nil
+	h.tickTolerant(12)
+	if h.gateway.admitted["org-a"] != "" || countGatewayCalls(h.gateway.calls, "open:") != 0 {
+		t.Fatal("missing tenant catalog reader allowed admission")
+	}
+}
+
+func TestTrinoUnpublishedTenantDoesNotBlockAnotherTenant(t *testing.T) {
+	h := newOperatorHarness(t)
+	h.operator.config.Pool.TenantAdmission = true
+	h.servingPool(t)
+	absent, ready := poolOrg("analyst"), poolOrg("engineer")
+	ready.OrgID, ready.DatabaseName = "org-b", "warehouse_b"
+	h.operator.tenants = &fakeTenantStore{orgs: []configstore.TrinoEnabledOrg{absent, ready}}
+	h.operator.catalogPublicationRevision = func(_ context.Context, catalog string) (int64, bool, error) {
+		if catalog == configstore.TrinoCatalogName(ready.DatabaseName) {
+			return 2, true, nil
+		}
+		return 0, false, nil
+	}
+	h.tick(t, 30)
+	if h.gateway.admitted[absent.OrgID] != "" || h.gateway.admitted[ready.OrgID] == "" {
+		t.Fatalf("admissions = %v, want only the tenant with a published catalog", h.gateway.admitted)
+	}
+}
+
+func TestTrinoAdmissionRechecksCatalogAfterLeaderRestart(t *testing.T) {
+	for _, opened := range []bool{false, true} {
+		t.Run(fmt.Sprintf("gateway_open=%t", opened), func(t *testing.T) {
+			h := newOperatorHarness(t)
+			h.operator.config.Pool.TenantAdmission = true
+			h.servingPool(t)
+			h.operator.tenants = &fakeTenantStore{orgs: []configstore.TrinoEnabledOrg{poolOrg("analyst")}}
+			for tick := 0; tick < 10 && countGatewayCalls(h.gateway.calls, "open:") == 0; tick++ {
+				h.tick(t, 1)
+			}
+			row := h.publications.rows["org-a"]
+			if row == nil || row.PublicationID == "" {
+				t.Fatal("expected an open admission attempt")
+			}
+			if !opened {
+				delete(h.gateway.publications, row.PublicationID)
+			}
+			successor := newOperatorHarness(t)
+			successor.store, successor.publications, successor.gateway = h.store, h.publications, h.gateway
+			successor.operator.store, successor.operator.publications, successor.operator.gateway = h.store, h.publications, h.gateway
+			successor.operator.tenants = h.operator.tenants
+			successor.operator.config.Pool.TenantAdmission = true
+			available := false
+			successor.operator.catalogPublicationRevision = func(context.Context, string) (int64, bool, error) {
+				if available {
+					return 2, true, nil
+				}
+				return 0, false, nil
+			}
+			successor.tick(t, 12)
+			if successor.gateway.admitted["org-a"] != "" || successor.publications.rows["org-a"].PublicationID != "" {
+				t.Fatal("successor reused admission evidence without a published tenant catalog")
+			}
+			available = true
+			successor.admitAll(t, 1)
+		})
+	}
+}
+
+func TestTrinoAdmissionUsesCatalogSnapshotNewerThanWatermark(t *testing.T) {
+	h := newOperatorHarness(t)
+	h.operator.config.Pool.TenantAdmission = true
+	h.servingPool(t)
+	h.operator.tenants = &fakeTenantStore{orgs: []configstore.TrinoEnabledOrg{poolOrg("analyst")}}
+	h.operator.catalogWatermark = func(context.Context) (int64, error) { return 1, nil }
+	h.operator.catalogPublicationRevision = func(context.Context, string) (int64, bool, error) { return 2, true, nil }
+	h.operator.acknowledgement = func(_ context.Context, _ string, _ trinoPoolProjectionRevisions, revision int64) (trinoPoolAcknowledgement, error) {
+		if revision != 2 {
+			t.Fatalf("required revision = %d, want the snapshot containing the tenant", revision)
+		}
+		return trinoPoolAcknowledgement{ProcessID: "process-1", AppliedRevision: 2, ProjectionCurrent: true}, nil
+	}
+	h.admitAll(t, 1)
+}
diff --git a/controlplane/trino_pool_catalog_hoglake_postgres_test.go b/controlplane/trino_pool_catalog_hoglake_postgres_test.go
index cbda9500..4eb541f7 100644
--- a/controlplane/trino_pool_catalog_hoglake_postgres_test.go
+++ b/controlplane/trino_pool_catalog_hoglake_postgres_test.go
@@ -133,6 +133,60 @@ func TestPooledCatalogWriterReportsPublishedConnectors(t *testing.T) {
 	}
 }
 
+func TestPooledCatalogWriterReadsTenantPublicationFromStore(t *testing.T) {
+	lease := configstore.TrinoPoolLease{PoolID: "registered:cell-001", Owner: "cp-test", Epoch: 1}
+	writer := scopedCatalogWriter(t, lease.PoolID, &recordingRevisionStore{}, lease)
+	ctx := context.Background()
+	if _, err := writer.db.ExecContext(ctx, `INSERT INTO trino_catalogs
+		(cell_id, catalog_name, connector_name, catalog_version, properties)
+		VALUES ($1, 'org_imported', 'hoglake', 'imported', '{}')`, writer.cellID); err != nil {
+		t.Fatal(err)
+	}
+	if err := writer.ClaimWriter(ctx); err != nil {
+		t.Fatal(err)
+	}
+	assertRevision := func(catalog string, want int64, wantPresent bool) {
+		t.Helper()
+		got, present, err := writer.CatalogPublicationRevision(ctx, catalog)
+		if err != nil || got != want || present != wantPresent {
+			t.Fatalf("catalog %s revision = %d, present=%t, err=%v; want %d, present=%t", catalog, got, present, err, want, wantPresent)
+		}
+	}
+	assertRevision("org_imported", 0, true)
+	assertRevision("org_first", 0, false)
+	properties := map[string]string{"connector.name": "hoglake", "hoglake.catalog": "example"}
+	if err := writer.CreateCatalog(ctx, "org_first", properties); err != nil {
+		t.Fatal(err)
+	}
+	assertRevision("org_first", 1, true)
+	assertRevision("org_second", 0, false)
+	if err := writer.CreateCatalog(ctx, "org_second", properties); err != nil {
+		t.Fatal(err)
+	}
+	assertRevision("org_first", 2, true)
+	assertRevision("org_second", 2, true)
+	if err := writer.DropCatalog(ctx, "org_first"); err != nil {
+		t.Fatal(err)
+	}
+	assertRevision("org_first", 0, false)
+
+	otherPartition := *writer
+	otherPartition.cellID = "another-pool-store"
+	if err := otherPartition.ClaimWriter(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := otherPartition.CreateCatalog(ctx, "org_first", properties); err != nil {
+		t.Fatal(err)
+	}
+	assertRevision("org_first", 0, false)
+
+	restarted := &trinoPoolCatalogWriter{db: writer.db, cellID: writer.cellID}
+	got, present, err := restarted.CatalogPublicationRevision(ctx, "org_second")
+	if err != nil || got != 3 || !present {
+		t.Fatalf("fresh writer revision = %d, %v; want the durable revision 3", got, err)
+	}
+}
+
 // The watermark the admission gate certifies against comes from the catalog
 // store, so a checkpoint that failed after a committed catalog is recoverable.
 //
diff --git a/controlplane/trino_pool_operator.go b/controlplane/trino_pool_operator.go
index 274e8598..a4bac3b7 100644
--- a/controlplane/trino_pool_operator.go
+++ b/controlplane/trino_pool_operator.go
@@ -162,6 +162,9 @@ type trinoPoolOperator struct {
 	// publishes through a coordinator, where the store is not duckgres-side and
 	// this question has no local answer.
 	catalogWatermark func(ctx context.Context) (int64, error)
+	// catalogPublicationRevision checks catalog presence and its store revision
+	// together; revision zero may contain catalogs imported before the writer.
+	catalogPublicationRevision func(ctx context.Context, catalog string) (int64, bool, error)
 	// bindingCursor and barrierCursor rotate which tenant is worked on. The
 	// driver performs one external step per tick, so a fixed order lets one
 	// permanently failing tenant hold the front of the queue forever - and with
diff --git a/controlplane/trino_pool_operator_test.go b/controlplane/trino_pool_operator_test.go
index a29c25f2..d3b231d7 100644
--- a/controlplane/trino_pool_operator_test.go
+++ b/controlplane/trino_pool_operator_test.go
@@ -824,6 +824,9 @@ func newOperatorHarness(t *testing.T) *operatorHarness {
 			},
 			identity:     func(context.Context, string) (string, error) { return "process-1", nil },
 			publications: publications,
+			catalogPublicationRevision: func(context.Context, string) (int64, bool, error) {
+				return store.pool.PublicationRevision, true, nil
+			},
 			// Every member is serving the projection the control plane is
 			// publishing. Tests that need the opposite override this.
 			acknowledgement: func(_ context.Context, _ string, _ trinoPoolProjectionRevisions, _ int64) (trinoPoolAcknowledgement, error) {
diff --git a/controlplane/trino_pool_publication.go b/controlplane/trino_pool_publication.go
index 653fb2f5..cfeeb98f 100644
--- a/controlplane/trino_pool_publication.go
+++ b/controlplane/trino_pool_publication.go
@@ -842,6 +842,10 @@ func isTrinoPoolCommittedPublication(err error) bool {
 // commit. Reusing an identity would resolve the previous attempt's recorded
 // outcome and leave this intent unapplied.
 func (o *trinoPoolOperator) openBarrier(ctx context.Context, binding trinoPoolTenantBinding) error {
+	basis, published, err := o.tenantBarrierBasis(ctx, binding)
+	if err != nil || !published {
+		return err
+	}
 	poolState, err := o.gateway.GetPool(ctx, o.config.RoutingGroup)
 	if err != nil {
 		return fmt.Errorf("read pool state for %s: %w", o.config.PublicID, err)
@@ -867,7 +871,7 @@ func (o *trinoPoolOperator) openBarrier(ctx context.Context, binding trinoPoolTe
 		o.config.PoolID, binding.Tenant, publicationID, target); err != nil {
 		return o.dropAuthority(fmt.Errorf("record publication intent for %s: %w", binding.Tenant, err))
 	}
-	o.rememberBarrierBasis(publicationID)
+	o.rememberBarrierBasis(publicationID, basis)
 	opened, err := o.issueOpenPublication(ctx, binding, target, publicationID, poolState.MembershipGeneration)
 	if err != nil {
 		return err
@@ -922,12 +926,19 @@ func (o *trinoPoolOperator) stepLiveBarrier(
 		}
 		// Recorded but absent on the Gateway: the open never landed. Re-issuing
 		// it under the SAME identity is the resolution, not a second barrier.
+		basis, published, err := o.tenantBarrierBasis(ctx, binding)
+		if err != nil {
+			return err
+		}
+		if !published {
+			return o.releaseBarrier(ctx, publication, "the tenant's catalog has not been published")
+		}
 		current, err = o.issueOpenPublication(ctx, binding, publication.TargetRevision,
 			publication.PublicationID, poolState.MembershipGeneration)
 		if err != nil {
 			return err
 		}
-		o.rememberBarrierBasis(publication.PublicationID)
+		o.rememberBarrierBasis(publication.PublicationID, basis)
 		return nil
 	}
 
@@ -1174,16 +1185,29 @@ func (o *trinoPoolOperator) acknowledge(
 	return acknowledgement, nil
 }
 
-// rememberBarrierBasis captures the configuration an attempt is admitting
-// against. There is at most one live attempt, so this holds one entry.
-func (o *trinoPoolOperator) rememberBarrierBasis(publicationID string) {
-	if o.barrierBasis == nil {
-		o.barrierBasis = map[string]trinoPoolBarrierBasis{}
+func (o *trinoPoolOperator) tenantBarrierBasis(ctx context.Context, binding trinoPoolTenantBinding) (trinoPoolBarrierBasis, bool, error) {
+	if o.catalogPublicationRevision == nil {
+		return trinoPoolBarrierBasis{}, false, fmt.Errorf("pool %s cannot verify tenant catalog publication", o.config.PublicID)
 	}
-	o.barrierBasis[publicationID] = trinoPoolBarrierBasis{
+	revision, present, err := o.catalogPublicationRevision(ctx, binding.Catalog)
+	if err != nil {
+		return trinoPoolBarrierBasis{}, false, err
+	}
+	if !present {
+		return trinoPoolBarrierBasis{}, false, nil
+	}
+	return trinoPoolBarrierBasis{
 		Projection:      o.expectedProjection(),
-		CatalogRevision: o.publishedCatalogRevision(),
+		CatalogRevision: max(revision, o.publishedCatalogRevision()),
+	}, true, nil
+}
+
+// There is at most one live attempt, so this holds one entry.
+func (o *trinoPoolOperator) rememberBarrierBasis(publicationID string, basis trinoPoolBarrierBasis) {
+	if o.barrierBasis == nil {
+		o.barrierBasis = map[string]trinoPoolBarrierBasis{}
 	}
+	o.barrierBasis[publicationID] = basis
 }
 
 func (o *trinoPoolOperator) barrierBasisFor(publicationID string) (trinoPoolBarrierBasis, bool) {
diff --git a/controlplane/trino_pool_wiring.go b/controlplane/trino_pool_wiring.go
index d95441fb..64df9f95 100644
--- a/controlplane/trino_pool_wiring.go
+++ b/controlplane/trino_pool_wiring.go
@@ -246,6 +246,7 @@ func buildTrinoPoolOperators(
 			// itself, so a checkpoint that failed after a committed catalog is
 			// recovered rather than waiting for a mutation that will never come.
 			operator.catalogWatermark = writer.PublishedRevision
+			operator.catalogPublicationRevision = writer.CatalogPublicationRevision
 			operator.installWriter = func(ctx context.Context, lease configstore.TrinoPoolLease) error {
 				held := lease
 				authority.Store(&held)
diff --git a/docs/runbooks/trino-hoglake-provisioning.md b/docs/runbooks/trino-hoglake-provisioning.md
index 777bc72a..71fbb3b8 100644
--- a/docs/runbooks/trino-hoglake-provisioning.md
+++ b/docs/runbooks/trino-hoglake-provisioning.md
@@ -78,10 +78,22 @@ single-file mount contract and configure Trino to load those files. If readiness
 reports a provider observation failure, check that provider's mount and the
 control plane's pod-exec access; changing tenant passwords will not repair it.
 
-Readiness also requires the existing authentication and cell gates. It verifies
-metadata and connector availability, but does not perform S3 writes. The smoke
-test below provides that verification. The administrative OPA grant permits
-connector inventory; it does not grant tenant data writes.
+Before publishing a new Trino catalog, the control plane assumes the tenant's
+storage role and verifies a small object write, read, and delete inside that
+tenant's configured Hoglake prefix. An unavailable role or denied storage
+operation keeps provisioning pending; the normal reconciliation retries it.
+The probe never uses the control plane's own S3 permissions as a fallback.
+It uses the reserved `.duckgres-readiness/` child prefix, with a five-second
+attempt deadline and an independent three-second cleanup deadline. Retries
+reuse a process-specific object key, including after an uncertain write.
+Existing catalogs do not repeat this initialization probe.
+
+Readiness also requires the existing authentication and cell gates. Tenant
+admission waits for a published catalog-store snapshot containing that tenant's
+catalog, then waits for the serving members to acknowledge that revision.
+The smoke test below still verifies the complete Trino query path. The
+administrative OPA grant permits connector inventory; it does not grant tenant
+data writes.
 
 Hoglake warehouse deprovisioning, organization deletion, and warehouse replacement
 are blocked, including for disabled Trino clients. These operations must wait for
@@ -106,10 +118,10 @@ or ownership mismatches explicitly; do not drop catalogs to force a switch.
    Apply their infrastructure and verify the service can maintain the reserved
    S3 prefix.
 3. Apply tenant-role policies granting each tenant access only to its own
-   Duckling-name child prefix. The Trino pod role must be allowed to assume that
-   role. Custom roles outside the managed composition require equivalent grants.
+   Duckling-name child prefix. Both the control-plane role and the Trino pod role must be allowed to assume
+   that role. Custom roles outside the managed composition require equivalent grants.
 4. Allow the control plane and Trino cells to reach the Hoglake REST service, and
-   allow Trino and Hoglake to reach the required storage services.
+   allow the control plane, Trino, and Hoglake to reach the required storage services.
 5. Verify healthy control plane and shared-pool admission. The pool controller
    creates its own validation tenant; no static canary credentials are required.
 6. Enable the managed configuration, create a dedicated pilot tenant, select its