fix(trino): recover rollouts and gate tenant readiness

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,21 +46,28 @@ 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")
 	}
 	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
 		}

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
+}