PostHog
diff --git a/‎controlplane/credential_refresh_scheduler.go‎
Lines changed: 6 additions & 0 deletions b/‎controlplane/credential_refresh_scheduler.go‎
Lines changed: 6 additions & 0 deletions
diff --git a/‎controlplane/multitenant.go‎
Lines changed: 0 additions & 6 deletions b/‎controlplane/multitenant.go‎
Lines changed: 0 additions & 6 deletions
diff --git a/‎controlplane/shared_worker_activator.go‎
Lines changed: 19 additions & 10 deletions b/‎controlplane/shared_worker_activator.go‎
Lines changed: 19 additions & 10 deletions
diff --git a/‎controlplane/shared_worker_activator_credentials_test.go‎
Lines changed: 148 additions & 0 deletions b/‎controlplane/shared_worker_activator_credentials_test.go‎
Lines changed: 148 additions & 0 deletions
diff --git a/‎controlplane/sts_broker.go‎
Lines changed: 109 additions & 7 deletions b/‎controlplane/sts_broker.go‎
Lines changed: 109 additions & 7 deletions
@@ -10,6 +10,12 @@ import (
 	"github.com/posthog/duckgres/controlplane/configstore"
 )
 
+// credentialRefreshLookahead and the broker's cache safety margin now live in
+// sts_broker.go alongside stsSessionDuration: all three derive from the
+// env-overridable session duration (DUCKGRES_STS_SESSION_DURATION), and the
+// former compile-time guard (margin strictly greater than lookahead) holds by
+// construction — stsCacheSafetyMargin = credentialRefreshLookahead + 5m.
+
 // credentialRefreshScheduleStore is the slice of the runtime store the
 // scheduler depends on. Defined narrowly so unit tests can fake it without
 // pulling in the full store.
 
@@ -360,12 +360,6 @@ func SetupMultiTenant(
 		refreshActivator.resolveDucklingStatus = resolveDucklingStatus
 	}
 
-	// Half the configured STS session duration: a worker due for refresh
-	// gets picked up well before its current session token actually goes
-	// stale, with a full half-life of slack to retry transient STS / RPC
-	// failures on subsequent ticks.
-	const credentialRefreshLookahead = stsSessionDuration / 2
-
 	// Per-CP scheduler — runs on every CP regardless of leader status, since
 	// each CP refreshes only the workers it owns (filtered by cpInstanceID in
 	// the SQL). Running this on the janitor leader only would leave workers
 
@@ -371,9 +371,11 @@ func (a *SharedWorkerActivator) BuildActivationRequest(ctx context.Context, org
 		}
 	}
 	if a.resolveDucklingStatus == nil || err != nil {
-		// Config-store warehouses use static creds (no STS), so expiresAt
-		// stays nil and the credential refresh scheduler skips them.
-		dl, err = a.buildDuckLakeConfigFromConfigStore(ctx, org.Warehouse)
+		// Static-cred config-store warehouses (secret-ref S3 credentials)
+		// return a nil expiresAt and the credential refresh scheduler skips
+		// them; the "aws" provider path brokers STS creds and returns their
+		// expiration so the scheduler keeps those workers fresh.
+		dl, expiresAt, err = a.buildDuckLakeConfigFromConfigStore(ctx, org.Warehouse)
 	}
 	if err != nil {
 		return TenantActivationPayload{}, err
@@ -505,10 +507,10 @@ func ducklingMetadataStoreAddress(status *provisioner.DucklingStatus, orgID stri
 
 // buildDuckLakeConfigFromConfigStore reads infrastructure details from the config store
 // and K8s Secrets. Used for non-Crossplane warehouses (manual seed, MinIO, etc.).
-func (a *SharedWorkerActivator) buildDuckLakeConfigFromConfigStore(ctx context.Context, warehouse *configstore.ManagedWarehouseConfig) (server.DuckLakeConfig, error) {
+func (a *SharedWorkerActivator) buildDuckLakeConfigFromConfigStore(ctx context.Context, warehouse *configstore.ManagedWarehouseConfig) (server.DuckLakeConfig, *time.Time, error) {
 	metadataPassword, err := a.readSecretValue(ctx, warehouse.MetadataStoreCredentials)
 	if err != nil {
-		return server.DuckLakeConfig{}, fmt.Errorf("metadata store credentials: %w", err)
+		return server.DuckLakeConfig{}, nil, fmt.Errorf("metadata store credentials: %w", err)
 	}
 
 	dl := server.DuckLakeConfig{
@@ -532,7 +534,7 @@ func (a *SharedWorkerActivator) buildDuckLakeConfigFromConfigStore(ctx context.C
 	case warehouse.S3Credentials.Name != "":
 		accessKey, secretKey, sessionToken, err := a.readS3Credentials(ctx, warehouse.S3Credentials)
 		if err != nil {
-			return server.DuckLakeConfig{}, fmt.Errorf("s3 credentials: %w", err)
+			return server.DuckLakeConfig{}, nil, fmt.Errorf("s3 credentials: %w", err)
 		}
 		dl.S3Provider = "config"
 		dl.S3AccessKey = accessKey
@@ -548,22 +550,29 @@ func (a *SharedWorkerActivator) buildDuckLakeConfigFromConfigStore(ctx context.C
 	case strings.EqualFold(warehouse.S3.Provider, "aws"):
 		roleARN := warehouse.WorkerIdentity.IAMRoleARN
 		if roleARN == "" {
-			return server.DuckLakeConfig{}, fmt.Errorf("managed warehouse %q requires worker_identity.iam_role_arn for worker activation", warehouse.OrgID)
+			return server.DuckLakeConfig{}, nil, fmt.Errorf("managed warehouse %q requires worker_identity.iam_role_arn for worker activation", warehouse.OrgID)
 		}
 		if a.stsBroker == nil {
-			return server.DuckLakeConfig{}, fmt.Errorf("STS broker is required for worker activation for org %q", warehouse.OrgID)
+			return server.DuckLakeConfig{}, nil, fmt.Errorf("STS broker is required for worker activation for org %q", warehouse.OrgID)
 		}
 		creds, err := a.stsBroker.AssumeRole(ctx, roleARN)
 		if err != nil {
-			return server.DuckLakeConfig{}, fmt.Errorf("STS AssumeRole for org %q: %w", warehouse.OrgID, err)
+			return server.DuckLakeConfig{}, nil, fmt.Errorf("STS AssumeRole for org %q: %w", warehouse.OrgID, err)
 		}
 		dl.S3Provider = "config"
 		dl.S3AccessKey = creds.AccessKeyID
 		dl.S3SecretKey = creds.SecretAccessKey
 		dl.S3SessionToken = creds.SessionToken
+		// STS-vended creds expire: surface the expiration so the
+		// credential-refresh scheduler keeps this worker's secret fresh.
+		// Without it the worker record's expiry stays NULL, the scheduler
+		// never lists the worker as due, and any session outliving the
+		// 1h token dies with ExpiredToken.
+		expiresAt := creds.Expiration
+		return dl, &expiresAt, nil
 	}
 
-	return dl, nil
+	return dl, nil, nil
 }
 
 func BuildTenantActivationPayload(ctx context.Context, clientset kubernetes.Interface, defaultNamespace string, org *configstore.OrgConfig, stsBroker *STSBroker) (TenantActivationPayload, error) {
 
@@ -0,0 +1,148 @@
+//go:build kubernetes
+
+package controlplane
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/posthog/duckgres/controlplane/configstore"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+// testWarehouseConfig returns a minimal config-store warehouse whose metadata
+// credentials resolve against the fake clientset's "metadata-creds" secret.
+func testWarehouseConfig() *configstore.ManagedWarehouseConfig {
+	return &configstore.ManagedWarehouseConfig{
+		OrgID: "org-test",
+		MetadataStore: configstore.ManagedWarehouseMetadataStore{
+			Endpoint:     "pg.test.local",
+			Port:         5432,
+			DatabaseName: "ducklake",
+			Username:     "duck",
+		},
+		MetadataStoreCredentials: configstore.SecretRef{
+			Namespace: "duckgres",
+			Name:      "metadata-creds",
+			Key:       "password",
+		},
+		S3: configstore.ManagedWarehouseS3{
+			Region: "us-east-1",
+			Bucket: "test-bucket",
+		},
+	}
+}
+
+func testMetadataSecret() *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: "metadata-creds", Namespace: "duckgres"},
+		Data:       map[string][]byte{"password": []byte("hunter2")},
+	}
+}
+
+// The "aws" provider path brokers STS credentials, and those expire: the
+// returned expiration is what the credential-refresh scheduler keys on to
+// keep the worker's DuckDB secret fresh. A nil expiry here means the
+// scheduler never refreshes the worker and any session outliving the STS
+// token dies with ExpiredToken mid-query.
+func TestBuildDuckLakeConfigFromConfigStoreAWSProviderReturnsExpiry(t *testing.T) {
+	expiration := time.Date(2026, 6, 11, 13, 0, 0, 0, time.UTC)
+	broker := newSTSBroker(func(ctx context.Context, roleARN string) (*AssumedCredentials, error) {
+		return &AssumedCredentials{
+			AccessKeyID:     "ASIATEST",
+			SecretAccessKey: "secret",
+			SessionToken:    "token",
+			Expiration:      expiration,
+		}, nil
+	})
+	a := &SharedWorkerActivator{
+		clientset:        fake.NewSimpleClientset(testMetadataSecret()),
+		defaultNamespace: "duckgres",
+		stsBroker:        broker,
+	}
+	warehouse := testWarehouseConfig()
+	warehouse.S3.Provider = "aws"
+	warehouse.WorkerIdentity = configstore.ManagedWarehouseWorkerIdentity{
+		IAMRoleARN: "arn:aws:iam::123:role/duckling-test",
+	}
+
+	dl, expiresAt, err := a.buildDuckLakeConfigFromConfigStore(context.Background(), warehouse)
+	if err != nil {
+		t.Fatalf("buildDuckLakeConfigFromConfigStore: %v", err)
+	}
+	if dl.S3AccessKey != "ASIATEST" || dl.S3SessionToken != "token" {
+		t.Errorf("expected STS-brokered creds on config, got access_key=%q session_token=%q", dl.S3AccessKey, dl.S3SessionToken)
+	}
+	if expiresAt == nil {
+		t.Fatal("expected non-nil credential expiry for STS-brokered aws provider; nil disables the credential-refresh scheduler for this worker")
+	}
+	if !expiresAt.Equal(expiration) {
+		t.Errorf("expiresAt = %v, want %v", *expiresAt, expiration)
+	}
+}
+
+// Static secret-ref credentials have no known expiration — the scheduler
+// must skip them, so the path must return a nil expiry.
+func TestBuildDuckLakeConfigFromConfigStoreStaticCredsNilExpiry(t *testing.T) {
+	s3Secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: "s3-creds", Namespace: "duckgres"},
+		Data: map[string][]byte{
+			"credentials": []byte(`{"access_key_id":"AKIATEST","secret_access_key":"secret"}`),
+		},
+	}
+	a := &SharedWorkerActivator{
+		clientset:        fake.NewSimpleClientset(testMetadataSecret(), s3Secret),
+		defaultNamespace: "duckgres",
+	}
+	warehouse := testWarehouseConfig()
+	warehouse.S3Credentials = configstore.SecretRef{
+		Namespace: "duckgres",
+		Name:      "s3-creds",
+		Key:       "credentials",
+	}
+
+	dl, expiresAt, err := a.buildDuckLakeConfigFromConfigStore(context.Background(), warehouse)
+	if err != nil {
+		t.Fatalf("buildDuckLakeConfigFromConfigStore: %v", err)
+	}
+	if dl.S3AccessKey != "AKIATEST" {
+		t.Errorf("expected static creds, got access_key=%q", dl.S3AccessKey)
+	}
+	if expiresAt != nil {
+		t.Errorf("static creds must return nil expiry (scheduler skip), got %v", *expiresAt)
+	}
+}
+
+// Pin the capture-point freshness floor: any credentials the broker hands
+// out (cached or fresh) must retain more validity than the refresh
+// scheduler's lookahead. If served creds could be closer to expiry than the
+// lookahead, a refresh push would stamp an expiry already inside the "due"
+// window and the scheduler would re-push identical creds every tick; worse,
+// a statement starting on those creds would have less runway than the
+// scheduler is designed to guarantee.
+func TestSTSBrokerServedCredsAlwaysOutliveSchedulerLookahead(t *testing.T) {
+	clock := &testClock{now: time.Date(2026, 6, 11, 12, 0, 0, 0, time.UTC)}
+	b := newTestBroker(clock, func(ctx context.Context, roleARN string) (*AssumedCredentials, error) {
+		return &AssumedCredentials{
+			AccessKeyID: "AKID",
+			Expiration:  clock.Now().Add(stsSessionDuration),
+		}, nil
+	})
+
+	// Sweep a full session duration in 1-minute steps; at every point the
+	// served creds must outlive the scheduler lookahead.
+	for i := 0; i <= int(stsSessionDuration/time.Minute); i++ {
+		creds, err := b.AssumeRole(context.Background(), "arn:aws:iam::123:role/org-a")
+		if err != nil {
+			t.Fatalf("AssumeRole at +%dm: %v", i, err)
+		}
+		if remaining := creds.Expiration.Sub(clock.Now()); remaining <= credentialRefreshLookahead {
+			t.Fatalf("at +%dm served creds have %v remaining, must exceed scheduler lookahead %v",
+				i, remaining, credentialRefreshLookahead)
+		}
+		clock.Advance(time.Minute)
+	}
+}
@@ -5,6 +5,9 @@ package controlplane
 import (
 	"context"
 	"fmt"
+	"log/slog"
+	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -15,21 +18,120 @@ import (
 )
 
 const (
-	stsSessionDuration = 1 * time.Hour
-	stsSessionName     = "duckgres-cp"
+	stsSessionName = "duckgres-cp"
 
-	// stsCacheSafetyMargin is how long before the credentials' true expiration
-	// we stop serving them from cache and mint a fresh session instead. It must
-	// leave enough room for a worker activation that receives cached creds to
-	// complete its DuckLake/Iceberg attach before the creds lapse.
-	stsCacheSafetyMargin = 10 * time.Minute
+	// defaultSTSSessionDuration is the AssumeRole DurationSeconds used unless
+	// overridden by the env-only DUCKGRES_STS_SESSION_DURATION knob.
+	defaultSTSSessionDuration = 1 * time.Hour
+
+	// minSTSSessionDuration is AWS's hard AssumeRole minimum (900s). The env
+	// knob is clamped up to this so a typo can't make AssumeRole reject every
+	// activation.
+	minSTSSessionDuration = 15 * time.Minute
+
+	// maxSTSSessionDuration is the effective AssumeRole ceiling in our
+	// deployment: the CP's own credentials come from EKS Pod Identity (a role
+	// session), so the per-org AssumeRole is role chaining, which STS
+	// hard-caps at 1h — and every duckling-* role's MaxSessionDuration is
+	// 3600s anyway. A DurationSeconds above that is not silently truncated:
+	// STS REJECTS the AssumeRole call, which would break activation for every
+	// org. The env knob is clamped down to this for the same reason it is
+	// clamped up to the minimum. Raising it for real requires de-chaining the
+	// assume AND raising MaxSessionDuration on every duckling role (12h
+	// absolute AWS max).
+	maxSTSSessionDuration = 1 * time.Hour
 
 	// stsAssumeRoleTimeout bounds the underlying AWS AssumeRole call. The call
 	// is detached from the triggering caller's context (other callers may be
 	// waiting on its result via singleflight), so it needs its own deadline.
 	stsAssumeRoleTimeout = 1 * time.Minute
 )
 
+// stsSessionDuration is the STS AssumeRole session duration. Env-overridable
+// (DUCKGRES_STS_SESSION_DURATION, e.g. "900s" / "15m" / "1h", clamped to
+// [minSTSSessionDuration, maxSTSSessionDuration]) so a soak test can shorten
+// tokens enough to exercise real in-statement expiry — with the production 1h
+// tokens, proving "a statement outlives its STS token" needs a >25min query
+// even with the refresh scheduler running. Only shortening is supported:
+// values above 1h would make AssumeRole itself fail (see
+// maxSTSSessionDuration).
+var stsSessionDuration = resolveSTSSessionDuration(os.Getenv("DUCKGRES_STS_SESSION_DURATION"))
+
+// credentialRefreshLookahead is how far ahead of a worker's recorded
+// credential expiry the scheduler refreshes it. Half the STS session
+// duration: a worker due for refresh gets picked up well before its current
+// session token actually goes stale, with a full half-life of slack to retry
+// transient STS / RPC failures on subsequent ticks.
+var credentialRefreshLookahead = stsSessionDuration / 2
+
+// stsCacheSafetyMargin is how long before the credentials' true expiration
+// we stop serving them from cache and mint a fresh session instead.
+//
+// This is the freshness floor for every credential capture point: all
+// AssumeRole callers bake the result into a worker's DuckDB secret, and a
+// DuckDB statement captures those credentials when it starts executing — a
+// later CREATE OR REPLACE SECRET (the credential-refresh scheduler's push)
+// does not re-credential an already-running statement (DuckDB resolves
+// secrets through the statement's MVCC snapshot). Stock httpfs has NO
+// mid-statement recovery path for scan workloads: its refresh-on-403 hook
+// only runs at file open AND only when the open performs a network request,
+// but DuckLake, Iceberg, and S3-glob scans all pre-populate file
+// size/etag/last-modified so opens skip the HEAD entirely and the first auth
+// failure surfaces on a range GET, which is not retried (verified against
+// httpfs v1.5.3 / ducklake / duckdb-iceberg sources; pinned by
+// TestInFlightScanDiesOnCredentialRotation in tests/integration). A
+// statement on stock httpfs therefore lives or dies on its starting runway:
+// this margin guarantees every statement at least lookahead+5m of token
+// validity at start; statements longer than that can still die with
+// ExpiredToken.
+//
+// The PostHog httpfs fork (PostHog/duckdb-httpfs, branch
+// cred-refresh-read-path) FIXES this: on an auth failure in any S3 request
+// path it re-resolves the latest committed secret — picking up this
+// scheduler's rotation pushes — and retries (proven by
+// TestInFlightScanSurvivesRotationWithPatchedHTTPFS in tests/integration).
+// Once a fork release with that patch is pinned via HTTPFS_EXTENSION_TAG in
+// Dockerfile.worker, this margin becomes defense-in-depth (a statement's
+// first credentials still want a healthy runway so recovery stays rare)
+// rather than the only thing standing between a long statement and
+// ExpiredToken.
+//
+// Defined as lookahead + 5m so it stays strictly greater than
+// credentialRefreshLookahead by construction (at the default 1h session:
+// 30m + 5m = 35m, the historical floor). If the margin were <= the
+// lookahead, a refresh push could stamp an expiry already inside the
+// scheduler's lookahead window, leaving the worker perpetually "due" and
+// re-pushing identical cached creds every tick until the margin finally
+// forces a fresh mint.
+var stsCacheSafetyMargin = credentialRefreshLookahead + 5*time.Minute
+
+// resolveSTSSessionDuration parses the env override, falling back to the
+// default on empty/garbage and clamping to AWS's AssumeRole minimum.
+func resolveSTSSessionDuration(raw string) time.Duration {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return defaultSTSSessionDuration
+	}
+	d, err := time.ParseDuration(raw)
+	if err != nil {
+		slog.Warn("Invalid DUCKGRES_STS_SESSION_DURATION; using default.",
+			"value", raw, "default", defaultSTSSessionDuration, "error", err)
+		return defaultSTSSessionDuration
+	}
+	if d < minSTSSessionDuration {
+		slog.Warn("DUCKGRES_STS_SESSION_DURATION below AWS AssumeRole minimum; clamping.",
+			"value", d, "minimum", minSTSSessionDuration)
+		return minSTSSessionDuration
+	}
+	if d > maxSTSSessionDuration {
+		slog.Warn("DUCKGRES_STS_SESSION_DURATION above the role-chaining AssumeRole ceiling; clamping. "+
+			"A larger DurationSeconds would make STS reject every per-org AssumeRole and break activation.",
+			"value", d, "maximum", maxSTSSessionDuration)
+		return maxSTSSessionDuration
+	}
+	return d
+}
+
 // assumeRoleFunc mints fresh credentials for a role ARN. It is a field on
 // STSBroker so tests can substitute a fake without an AWS client.
 type assumeRoleFunc func(ctx context.Context, roleARN string) (*AssumedCredentials, error)