Skip to content

test: add archiver crash rollback and partitioner FK drop rollback journey stories #45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
imtiazqa wants to merge 2 commits into
base: main
Choose a base branch
from
Open

test: add archiver crash rollback and partitioner FK drop rollback journey stories #45

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1dcce2d
test: add archiver crash rollback and partitioner FK drop rollback st…
Jul 8, 2026
4507e07
test: skip partitioner FK drop rollback in mesh mode (peer propagation)
Jul 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 169 additions & 0 deletions ci/journey.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1704,6 +1704,173 @@ EOF
q "$HOST" "DROP TABLE IF EXISTS public.fk_categories CASCADE;" >/dev/null 2>&1
}

# ───────────────────────────────────────────────────────────────────────────
# Story — archiver mid-archive crash rollback: kill the archiver during the
# post-Phase-2 debug window and verify that on re-run Phase 0 wipes the
# partial Iceberg data and the archive completes cleanly. Asserts no data is
# lost from PostgreSQL during a crash.
# ───────────────────────────────────────────────────────────────────────────
story_archiver_rollback() {
step "Archiver rollback: mid-archive crash → Phase 0 self-heals on re-run"

local dsn="host=${DB_IP} port=5432 dbname=coldfront user=coldfront password=coldfront sslmode=disable"
local ret_days; ret_days=$(( ( $(date -u +%s) - $(date -u -d "$(date -u +%Y-%m-01) -1 month" +%s) ) / 86400 ))

# Partitioned table with one cold partition (3 months ago) and one hot partition (current month).
qf "$HOST" <<'EOSQL' >/dev/null
SET search_path = public;
CREATE TABLE IF NOT EXISTS public.rb_events (
id bigint GENERATED ALWAYS AS IDENTITY,
ts timestamptz NOT NULL,
status text,
PRIMARY KEY (id, ts)
) PARTITION BY RANGE (ts);
DO $do$
DECLARE m date;
BEGIN
m := date_trunc('month', now()) - interval '3 months';
EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF rb_events FOR VALUES FROM (%L) TO (%L)',
'rb_events_p_cold', m, m + interval '1 month');
m := date_trunc('month', now());
EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF rb_events FOR VALUES FROM (%L) TO (%L)',
'rb_events_p_hot', m, m + interval '1 month');
END $do$;
EOSQL

q "$HOST" "INSERT INTO public.rb_events (ts, status) VALUES (date_trunc('month',now()) - interval '3 months' + interval '1 day', 'crash-row'), (now(), 'hot-row');" >/dev/null

if "$ARCHIVER" register --config /tmp/journey-archiver.yaml --table rb_events \
--period monthly --hot-period "${ret_days} days" >/tmp/journey-rb-reg.log 2>&1; then
pass "rb_events registered"
else
fail "rb_events register failed"; tail -5 /tmp/journey-rb-reg.log; return
fi

cat > /tmp/journey-rb.yaml <<EOF
postgres: { dsn: "${dsn}" }
iceberg: { warehouse: "${WAREHOUSE}", lakekeeper_endpoint: "http://${LK_IP}:8181/catalog", namespace: "default" }
$(storage_yaml)
EOF

# Run with a 60s debug delay so Phase 2 completes before we kill.
"$ARCHIVER" --config /tmp/journey-rb.yaml --debug-export-delay 60s >/tmp/journey-rb-crash.log 2>&1 &
local arch_pid=$!
sleep 10
kill "$arch_pid" 2>/dev/null || true
wait "$arch_pid" 2>/dev/null || true

# Data must still be in PG — not lost.
assert_eq "crash-row still in PG after mid-archive crash" "1" \
"$(q "$HOST" "SELECT count(*) FROM public.rb_events WHERE status='crash-row';")"

# Re-run: Phase 0 wipes partial Iceberg data; full archive must complete.
if "$ARCHIVER" --config /tmp/journey-rb.yaml >/tmp/journey-rb-rerun.log 2>&1; then
pass "archiver self-healed on re-run (Phase 0 wiped partial Iceberg data)"
else
fail "archiver re-run failed after crash"; tail -8 /tmp/journey-rb-rerun.log; return
fi

assert_eq "watermark written after successful re-run" "1" \
"$(q "$HOST" "SELECT count(*) FROM coldfront.archive_watermark WHERE table_name='rb_events';")"
assert_eq "both rows visible via unified view after re-run" "2" \
"$(q "$HOST" "SELECT count(*) FROM public.rb_events;")"

# Cleanup.
q "$HOST" "DELETE FROM coldfront.partition_config WHERE table_name='rb_events';" >/dev/null 2>&1
q "$HOST" "DELETE FROM coldfront.archive_watermark WHERE table_name='rb_events';" >/dev/null 2>&1
q "$HOST" "DROP VIEW IF EXISTS public.rb_events CASCADE;" >/dev/null 2>&1
q "$HOST" "DROP TABLE IF EXISTS public._rb_events CASCADE;" >/dev/null 2>&1
q "$HOST" "DROP TABLE IF EXISTS public.rb_events CASCADE;" >/dev/null 2>&1
}

# ───────────────────────────────────────────────────────────────────────────
# Story — partitioner inbound FK blocks expired partition drop: asserts the
# partitioner fails fast (SQLSTATE 23503, no retry), the partition remains
# attached, and on re-run after the FK reference is removed the partition is
# dropped cleanly.
# ───────────────────────────────────────────────────────────────────────────
story_partitioner_fk_drop() {
step "Partitioner rollback: inbound FK blocks expired partition drop → self-heals after FK removed"

# Mesh topology propagates partition detach to peers; peer DNS is not
# available in the single-node journey config, so skip in mesh mode.
if [ "$MESH" = 1 ]; then
note "skipping partitioner FK drop rollback in mesh mode (peer propagation requires mesh config)"
return
fi

local dsn="host=${DB_IP} port=5432 dbname=coldfront user=coldfront password=coldfront sslmode=disable"
printf 'postgres: { dsn: "%s" }\n' "$dsn" > /tmp/journey-pfk.yaml

# Partitioned table with one expired partition (3 months old) and one current partition.
qf "$HOST" <<'EOSQL' >/dev/null
SET search_path = public;
CREATE TABLE IF NOT EXISTS public.pfk_logs (
id bigint GENERATED ALWAYS AS IDENTITY,
ts timestamptz NOT NULL,
msg text,
PRIMARY KEY (id, ts)
) PARTITION BY RANGE (ts);
DO $do$
DECLARE m date;
BEGIN
m := date_trunc('month', now()) - interval '3 months';
EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF pfk_logs FOR VALUES FROM (%L) TO (%L)',
'pfk_logs_p_expired', m, m + interval '1 month');
m := date_trunc('month', now());
EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF pfk_logs FOR VALUES FROM (%L) TO (%L)',
'pfk_logs_p_current', m, m + interval '1 month');
END $do$;
EOSQL

q "$HOST" "INSERT INTO public.pfk_logs (ts, msg) VALUES (date_trunc('month',now()) - interval '3 months' + interval '1 day', 'old-row'), (now(), 'hot-row');" >/dev/null

# Inbound FK from pfk_log_refs → pfk_logs(id, ts) references the expired row.
qf "$HOST" <<'EOSQL' >/dev/null
CREATE TABLE IF NOT EXISTS public.pfk_log_refs (
ref_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
log_id bigint NOT NULL,
log_ts timestamptz NOT NULL,
FOREIGN KEY (log_id, log_ts) REFERENCES public.pfk_logs(id, ts)
);
INSERT INTO public.pfk_log_refs (log_id, log_ts)
SELECT id, ts FROM public.pfk_logs WHERE msg = 'old-row';
EOSQL

if "$PARTITIONER" register --dsn "$dsn" --schema public --table pfk_logs \
--period monthly --retention "1 month" >/tmp/journey-pfk-reg.log 2>&1; then
pass "pfk_logs registered as partition-only"
else
fail "pfk_logs register failed"; tail -5 /tmp/journey-pfk-reg.log; return
fi

# Run partitioner — inbound FK must block the expired partition drop and fail fast.
local pfk_out
pfk_out=$("$PARTITIONER" --config /tmp/journey-pfk.yaml 2>&1 || true)
assert_contains "partitioner fails fast on inbound FK (SQLSTATE 23503)" "23503" "$pfk_out"

# Expired partition must still be attached — data safe in PG.
assert_eq "expired partition still attached after FK-blocked drop" "1" \
"$(q "$HOST" "SELECT count(*) FROM pg_class WHERE relname='pfk_logs_p_expired';")"

# Fix: remove the referencing row then re-run.
q "$HOST" "DELETE FROM public.pfk_log_refs WHERE log_ts < date_trunc('month',now()) - interval '2 months';" >/dev/null

if "$PARTITIONER" --config /tmp/journey-pfk.yaml >/tmp/journey-pfk-rerun.log 2>&1; then
pass "partitioner self-healed after FK reference removed"
else
fail "partitioner re-run failed after FK removed"; tail -8 /tmp/journey-pfk-rerun.log; return
fi

assert_eq "expired partition dropped after FK removed" "0" \
"$(q "$HOST" "SELECT count(*) FROM pg_class WHERE relname='pfk_logs_p_expired';")"

# Cleanup.
q "$HOST" "DELETE FROM coldfront.partition_config WHERE table_name='pfk_logs';" >/dev/null 2>&1
q "$HOST" "DROP TABLE IF EXISTS public.pfk_log_refs CASCADE;" >/dev/null 2>&1
q "$HOST" "DROP TABLE IF EXISTS public.pfk_logs CASCADE;" >/dev/null 2>&1
}

# ───────────────────────────────────────────────────────────────────────────
# Story — management CLI end-to-end: register a table via the subcommand (PK
# validation), list it, reject a PK-less table, then run the archiver with NO
Expand Down Expand Up @@ -2111,6 +2278,8 @@ if [ "$MODE" = "tiered" ]; then
story_partitioner_multitable
story_partitioner_after_swap
story_fk_constraint
story_archiver_rollback
story_partitioner_fk_drop
story_register_cli
story_partition_config_ownership # each binary loads only its own rows
else
Expand Down