diff --git a/Cargo.lock b/Cargo.lock index ffab16b..01ce396 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3474,6 +3474,7 @@ dependencies = [ "serde_json", "thiserror", "tokio", + "tracing", "tracing-subscriber", ] diff --git a/ivy_moonlink b/ivy_moonlink index 9a5af42..b42163a 160000 --- a/ivy_moonlink +++ b/ivy_moonlink @@ -1 +1 @@ -Subproject commit 9a5af42ad1f1488afa25497c8f145c211a28d7fa +Subproject commit b42163a2353d51508e3f6dfb247c2c7a59df15e9 diff --git a/src/functions.rs b/src/functions.rs index 63625e8..8f53f6b 100644 --- a/src/functions.rs +++ b/src/functions.rs @@ -211,7 +211,22 @@ fn uri_encode(input: &str) -> String { /// then fail to resolve in the CREATE TABLE. Propagate the calling /// session's mode so DDL built from Oracle-typed tables works. fn propagate_compatible_mode(client: &mut Client) { - let mode = Spi::get_one::("SELECT current_setting('ivorysql.compatible_mode', true)"); + // Read the GUC via the read-only SPI path. `Spi::get_one` routes through + // `SpiClient::update`, whose `mark_mutable` assigns the calling backend a + // real transaction id. That xid deadlocks create_table: the moonlink RPC + // issued later blocks on CREATE_REPLICATION_SLOT ... USE_SNAPSHOT, whose + // snapshot builder waits for every in-progress write xact — including + // ours — while we sit in block_on waiting for that same RPC to return. + let mode = Spi::connect(|client| { + client + .select( + "SELECT current_setting('ivorysql.compatible_mode', true)", + Some(1), + &[], + )? + .first() + .get_one::() + }); if let Ok(Some(mode)) = mode { if mode == "oracle" { client diff --git a/src/sql/bootstrap.sql b/src/sql/bootstrap.sql index 54fa024..99fae1b 100644 --- a/src/sql/bootstrap.sql +++ b/src/sql/bootstrap.sql @@ -12,16 +12,18 @@ SELECT duckdb.install_extension('mooncake', 'community'); CREATE FUNCTION public.mooncake_extension_drop_cleanup() RETURNS event_trigger LANGUAGE plpgsql AS $mooncake_drop_cleanup$ DECLARE - is_mooncake_drop boolean := false; slot record; BEGIN -- Act only when pg_mooncake itself is among the dropped objects. - SELECT true INTO is_mooncake_drop - FROM pg_event_trigger_dropped_objects() - WHERE object_type = 'extension' AND object_name = 'pg_mooncake' - LIMIT 1; - - IF NOT is_mooncake_drop THEN + -- Use EXISTS, not SELECT INTO: SELECT INTO with zero rows overwrites the + -- variable with NULL (even when initialized false), and IF NOT NULL is + -- not taken — so the cleanup body would run on EVERY sql_drop, spuriously + -- dropping the slot/publication and self-removing on any DROP TABLE. + IF NOT EXISTS ( + SELECT 1 + FROM pg_event_trigger_dropped_objects() + WHERE object_type = 'extension' AND object_name = 'pg_mooncake' + ) THEN RETURN; END IF; diff --git a/src/utils.rs b/src/utils.rs index 09116b2..463981f 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -22,10 +22,25 @@ pub(crate) fn block_on(future: F) -> F::Output { pub(crate) fn get_stream() -> MutexGuard<'static, UnixStream> { static STREAM: LazyLock> = LazyLock::new(|| { - Mutex::new( - block_on(UnixStream::connect("pg_mooncake/moonlink.sock")) - .expect("Failed to connect to moonlink"), - ) + // The moonlink bgworker binds its socket asynchronously after postmaster + // start, so the first client call in a fresh cluster can race it and see + // ECONNREFUSED/ENOENT. Retry with backoff instead of failing the backend: + // a panic here would also poison this LazyLock, permanently breaking every + // later mooncake call in the session. + let mut delay = std::time::Duration::from_millis(50); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + let stream = loop { + match block_on(UnixStream::connect("pg_mooncake/moonlink.sock")) { + Ok(stream) => break stream, + Err(err) if std::time::Instant::now() < deadline => { + pgrx::log!("moonlink not ready ({err}), retrying in {delay:?}"); + std::thread::sleep(delay); + delay = (delay * 2).min(std::time::Duration::from_secs(1)); + } + Err(err) => panic!("Failed to connect to moonlink: {err:?}"), + } + }; + Mutex::new(stream) }); STREAM.lock().unwrap() }