Adopt the partitioner's tables at install, and test the reaper's epoch rule

ci/journey.sh

@@ -2360,6 +2360,37 @@ SQL
     wait "$holder" 2>/dev/null
     assert_eq "TC-163 peer deferred behind an orphan is unwedged by its own poke once the lock drops" "0" "$rc"
     assert_eq "TC-163 orphan claim reaped on both nodes" "0" "$(orphan_gone_everywhere)"
+
+    # TC-191: epoch path. A same-node claim on another table whose ticket predates
+    # this postmaster is a leftover of a previous owner: a restart, or a
+    # reassigned snowflake.node. The other-table rule cannot reap it while a live
+    # session holds that table's lock, so only the epoch rule does. The lock is
+    # held for the whole write; a write that outlived it would prove nothing.
+    local stale_tbl="${ref}_tc191" stale
+    stale=$(q "$HOST" "WITH ins AS (INSERT INTO coldfront.claims (iceberg_table, ticket) SELECT '$stale_tbl', snowflake.nextval() - (((extract(epoch FROM now() - pg_postmaster_start_time())::bigint + 3600) * 1000) << 22) RETURNING ticket) SELECT ticket FROM ins;")
+    assert_eq "TC-191 the planted ticket carries this node and predates the postmaster" "t" \
+        "$(q "$HOST" "SELECT snowflake.get_node($stale) = coldfront.node_id() AND snowflake.get_epoch($stale) < extract(epoch FROM pg_postmaster_start_time());")"
+    docker exec -i -e PGUSER="$CF_DBUSER" -e PGDATABASE="$CF_DBNAME" "$HOST" "$CF_PSQL" -tA -v ON_ERROR_STOP=1 >/dev/null 2>&1 <<SQL &
+BEGIN;
+SELECT pg_advisory_xact_lock(hashtext('coldfront_iceberg:' || '$stale_tbl'));
+SELECT pg_sleep(8);
+COMMIT;
+SQL
+    holder=$!
+    local w_d probe="SELECT pg_try_advisory_xact_lock(hashtext('coldfront_iceberg:' || '$stale_tbl'));"
+    if [ "$MODE" = tiered ]; then
+        w_d="INSERT INTO events (ts,status,data) VALUES (date_trunc('month',now()) - interval '4 months' + interval '23 days','reap191','{}');"
+    else
+        w_d="INSERT INTO iceonly VALUES (9104,date_trunc('month',now()) + interval '3 months' + interval '4 days','reap191','{}');"
+    fi
+    sleep 1
+    assert_eq "TC-191 the stale claim's table lock is held before the write" "f" "$(q "$HOST" "$probe")"
+    tq 90 "$HOST" "$w_d" >"$TMPD/reap.d" 2>&1; rc=$?
+    assert_eq "TC-191 the cold write to another table lands" "0" "$rc"
+    assert_eq "TC-191 the stale claim's table lock is still held after the write" "f" "$(q "$HOST" "$probe")"
+    wait "$holder" 2>/dev/null
+    assert_eq "TC-191 the stale claim was reaped by the epoch rule" "0" \
+        "$(q "$HOST" "SELECT count(*) FROM coldfront.claims WHERE ticket = $stale;")"
     assert_eq "TC-160..163 all three reaper-path writes landed" "3" \
         "$(q "$HOST" "SELECT count(*) FROM $tbl WHERE status='reap';")"
     rm -f $TMPD/reap.* 2>/dev/null
@@ -4112,6 +4143,17 @@ story_partitioner_stock_pg() {
         fail "TC-151: reconcile failed on stock PG"; tail -5 $TMPD/stock-run.log
     fi
 
+    # The cold tier can be added to this database later: CREATE EXTENSION adopts
+    # the partition_config the partitioner made, rows and all, instead of
+    # refusing to skip a table it does not own.
+    if qdb $db "CREATE EXTENSION pg_duckdb; CREATE EXTENSION coldfront;" >$TMPD/stock-ext.log 2>&1; then
+        pass "TC-151: the extension installs on a partitioner-first database"
+    else
+        fail "TC-151: CREATE EXTENSION failed on a partitioner-first database"; tail -3 $TMPD/stock-ext.log
+    fi
+    assert_eq "TC-151: partition_config is a member of the extension and kept its row" "1" \
+        "$(qdb $db "SELECT count(*) FROM coldfront.partition_config WHERE table_name = 'stockev' AND EXISTS (SELECT 1 FROM pg_depend d JOIN pg_extension e ON e.oid = d.refobjid WHERE d.objid = 'coldfront.partition_config'::regclass AND e.extname = 'coldfront' AND d.deptype = 'e');")"
+
     "$PARTITIONER" remove --dsn "$dsn" --table stockev >/dev/null 2>&1
     assert_eq "TC-151: remove unregistered it" "0" \
         "$(qdb $db "SELECT count(*) FROM coldfront.partition_config WHERE table_name='stockev';")"

docs/changelog.md

@@ -45,6 +45,9 @@ and this project adheres to
 
 ### Fixed
 
+- `CREATE EXTENSION coldfront` failed on a database the standalone partitioner
+  had already set up ("table partition_config is not a member of extension"):
+  the extension now adopts that table, registrations included.
 - After one cold write on a mesh, an app role could run any SQL as the
   loopback connection's user through the `coldfront_self` dblink connection
   the claim left open in its session, and by setting

docs/usage.md

@@ -307,6 +307,10 @@ go build -o bin/partitioner ./cmd/partitioner   # or: make build
 ./bin/partitioner --config config.yaml
 ```
 
+A database set up this way can gain the cold tier later: `CREATE EXTENSION
+coldfront` adopts the `coldfront.partition_config` the partitioner created,
+registrations included.
+
 A partition-only config omits the `iceberg:` and `s3:` sections entirely
 (supply either all of them or none - a half-filled cold config is
 rejected):

extension/coldfront/coldfront--1.0.sql

@@ -60,6 +60,28 @@ CREATE TABLE coldfront.tiered_views (
 -- ON CONFLICT (schema_name, table_name) DO UPDATE (internal/watermark/watermark.go:Set).
 -- Created with IF NOT EXISTS because the archiver can also materialize it
 -- on first run before CREATE EXTENSION runs against the DB.
+--
+-- A database the standalone partitioner set up before the extension holds
+-- coldfront.partition_config as a plain table, and one the archiver ran on
+-- holds archive_watermark the same way, both from the Go copies of this DDL
+-- (internal/partcfg, internal/watermark). An extension script may skip
+-- CREATE ... IF NOT EXISTS only for an object the extension owns, so such a
+-- table is adopted first: its rows stay, and the pg_extension_config_dump
+-- calls below then cover it like a table this script created.
+DO $$
+DECLARE t text;
+BEGIN
+    FOREACH t IN ARRAY ARRAY['archive_watermark', 'partition_config'] LOOP
+        IF to_regclass('coldfront.' || t) IS NOT NULL AND NOT EXISTS (
+               SELECT 1 FROM pg_depend
+                WHERE classid = 'pg_class'::regclass
+                  AND objid = to_regclass('coldfront.' || t)
+                  AND refclassid = 'pg_extension'::regclass
+                  AND deptype = 'e') THEN
+            EXECUTE format('ALTER EXTENSION coldfront ADD TABLE coldfront.%I', t);
+        END IF;
+    END LOOP;
+END $$;
 CREATE TABLE IF NOT EXISTS coldfront.archive_watermark (
     schema_name text        NOT NULL,
     table_name  text        NOT NULL,

internal/partcfg/mirror_test.go

@@ -0,0 +1,50 @@
+package partcfg
+
+import (
+	"os"
+	"regexp"
+	"strings"
+	"testing"
+
+	"github.com/pgedge/coldfront/internal/watermark"
+)
+
+// extensionTableDDL returns the CREATE TABLE block for one table from the
+// extension's install script, without its trailing comments and semicolon.
+func extensionTableDDL(t *testing.T, table string) string {
+	t.Helper()
+	script, err := os.ReadFile("../../extension/coldfront/coldfront--1.0.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	head := "CREATE TABLE IF NOT EXISTS coldfront." + table + " ("
+	start := strings.Index(string(script), head)
+	if start < 0 {
+		t.Fatalf("%s not found in the extension script", head)
+	}
+	end := strings.Index(string(script)[start:], "\n);")
+	if end < 0 {
+		t.Fatalf("%s: no closing line in the extension script", head)
+	}
+	return regexp.MustCompile(`--[^\n]*`).ReplaceAllString(string(script)[start:start+end+2], "")
+}
+
+// tokens collapses whitespace so two spellings of one DDL compare equal.
+func tokens(ddl string) string {
+	return strings.Join(strings.Fields(ddl), " ")
+}
+
+// The Go copies of the two tables the extension owns are what a database gets
+// when the partitioner or the archiver runs before CREATE EXTENSION, and the
+// extension adopts those tables as its own at install. So each copy has to be
+// the extension's DDL, token for token.
+func TestMirrorDDL_MatchesExtension(t *testing.T) {
+	for table, mirror := range map[string]string{
+		"partition_config":  CreateTableSQL,
+		"archive_watermark": watermark.CreateTableSQL,
+	} {
+		if got, want := tokens(mirror), tokens(extensionTableDDL(t, table)); got != want {
+			t.Errorf("%s: the Go copy differs from the extension script\n got: %s\nwant: %s", table, got, want)
+		}
+	}
+}

internal/partcfg/partcfg.go

@@ -25,11 +25,12 @@ type DBTX interface {
 	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
 }
 
-// createTableSQL mirrors the coldfront.partition_config DDL in the C extension
+// CreateTableSQL mirrors the coldfront.partition_config DDL in the C extension
 // (coldfront--1.0.sql) so the vanilla partitioner — stock PG, no extension —
-// can self-materialize the same table. Keep the two in sync (same pattern as
-// archive_watermark / watermark.EnsureTable).
-const createTableSQL = `
+// can self-materialize the same table. The extension adopts that table when it
+// is installed later, so the two are kept token-identical (TestMirrorDDL_MatchesExtension;
+// same pattern as archive_watermark / watermark.EnsureTable).
+const CreateTableSQL = `
 CREATE TABLE IF NOT EXISTS coldfront.partition_config (
     schema_name            text    NOT NULL DEFAULT 'public',
     table_name             text    NOT NULL,
@@ -92,7 +93,7 @@ func EnsureTable(ctx context.Context, db DBTX) error {
 	if _, err := db.Exec(ctx, `CREATE SCHEMA IF NOT EXISTS coldfront`); err != nil {
 		return fmt.Errorf("create schema: %w", err)
 	}
-	if _, err := db.Exec(ctx, createTableSQL); err != nil {
+	if _, err := db.Exec(ctx, CreateTableSQL); err != nil {
 		return fmt.Errorf("create partition_config: %w", err)
 	}
 	if _, err := db.Exec(ctx, addColumnsSQL); err != nil {

internal/watermark/watermark.go

@@ -26,18 +26,25 @@ func NewStore(db DBTX) *Store {
 	return &Store{db: db}
 }
 
+// CreateTableSQL mirrors the coldfront.archive_watermark DDL in the C extension
+// (coldfront--1.0.sql), so an archiver run can materialize the table before
+// CREATE EXTENSION. The extension adopts that table when it is installed
+// later, so the two are kept token-identical (partcfg's
+// TestMirrorDDL_MatchesExtension).
+const CreateTableSQL = `
+CREATE TABLE IF NOT EXISTS coldfront.archive_watermark (
+    schema_name text        NOT NULL,
+    table_name  text        NOT NULL,
+    cutoff_time timestamptz NOT NULL,
+    PRIMARY KEY (schema_name, table_name)
+)`
+
 // EnsureTable creates the coldfront schema and archive_watermark table if they don't exist.
 func (s *Store) EnsureTable(ctx context.Context) error {
 	if _, err := s.db.Exec(ctx, `CREATE SCHEMA IF NOT EXISTS coldfront`); err != nil {
 		return fmt.Errorf("create schema: %w", err)
 	}
-	if _, err := s.db.Exec(ctx, `
-		CREATE TABLE IF NOT EXISTS coldfront.archive_watermark (
-			schema_name text        NOT NULL,
-			table_name  text        NOT NULL,
-			cutoff_time timestamptz NOT NULL,
-			PRIMARY KEY (schema_name, table_name)
-		)`); err != nil {
+	if _, err := s.db.Exec(ctx, CreateTableSQL); err != nil {
 		return fmt.Errorf("create watermark table: %w", err)
 	}
 	return nil