sql-150: Create test suite for durable temporary objects

src/catalog/tests/open.rs

@@ -19,7 +19,8 @@ use mz_catalog::durable::objects::{DurableType, Snapshot};
 use mz_catalog::durable::{
     BUILTIN_MIGRATION_SHARD_KEY, CATALOG_VERSION, CatalogError, Database, DurableCatalogError,
     DurableCatalogState, EXPRESSION_CACHE_SHARD_KEY, Epoch, FenceError,
-    MOCK_AUTHENTICATION_NONCE_KEY, Schema, TestCatalogStateBuilder, test_bootstrap_args,
+    MOCK_AUTHENTICATION_NONCE_KEY, Schema, TestCatalogStateBuilder, Transaction,
+    test_bootstrap_args,
 };
 use mz_catalog_protos::objects::{SettingKey, SettingValue};
 use mz_ore::cast::usize_to_u64;
@@ -30,7 +31,9 @@ use mz_persist_client::{PersistClient, PersistLocation};
 use mz_persist_types::ShardId;
 use mz_proto::RustType;
 use mz_repr::role_id::RoleId;
+use mz_repr::{CatalogItemId, GlobalId};
 use mz_sql::catalog::{RoleAttributesRaw, RoleMembership, RoleVars};
+use mz_sql::names::SchemaId;
 use uuid::Uuid;
 
 /// A new type for [`Snapshot`] that excludes fields that change often from the debug output. It's
@@ -482,6 +485,295 @@ async fn test_open_read_only(state_builder: TestCatalogStateBuilder) {
     Box::new(state).expire().await;
 }
 
+/// The ids of all items in a catalog snapshot.
+fn item_ids(snapshot: &Snapshot) -> Vec<CatalogItemId> {
+    snapshot
+        .items
+        .keys()
+        .map(|key| CatalogItemId::from_proto(key.gid.clone()).unwrap())
+        .collect()
+}
+
+/// Inserts a view item. A `Some` owner session tags the item as temporary.
+fn insert_view(
+    txn: &mut Transaction<'_>,
+    id: CatalogItemId,
+    schema_id: SchemaId,
+    name: &str,
+    ephemeral_owner_session: Option<Uuid>,
+) {
+    let CatalogItemId::User(raw_id) = id else {
+        panic!("tests only insert user items");
+    };
+    txn.insert_item(
+        id,
+        u32::try_from(20_000 + raw_id).expect("small"),
+        GlobalId::User(raw_id),
+        schema_id,
+        name,
+        format!("CREATE VIEW {name} AS SELECT 1"),
+        RoleId::User(1),
+        vec![],
+        BTreeMap::new(),
+        ephemeral_owner_session,
+    )
+    .unwrap();
+}
+
+#[mz_ore::test(tokio::test)]
+#[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
+async fn test_persist_open_reclaims_ephemeral_items() {
+    let persist_client = PersistClient::new_for_tests().await;
+    let state_builder = TestCatalogStateBuilder::new(persist_client);
+    test_open_reclaims_ephemeral_items(state_builder).await;
+}
+
+/// Temporary items are durable, tagged with the UUID of the session that
+/// created them, so a process that dies without running its session-close
+/// cleanup leaves them behind. Opening the catalog with write intent fences out
+/// every previous owner, which means every session that could own one is dead,
+/// so that open reclaims them. This is the only thing standing between a
+/// `kill -9` and a permanently leaked catalog item.
+///
+/// A read-only open must not reclaim anything: during a zero-downtime deploy the
+/// follower reads the leader's catalog while the leader's sessions are still
+/// live and still own their temporary items.
+async fn test_open_reclaims_ephemeral_items(state_builder: TestCatalogStateBuilder) {
+    let state_builder = state_builder.with_default_deploy_generation();
+    let owner_session = Uuid::from_u128(1);
+    let ephemeral_id = CatalogItemId::User(200);
+    let normal_id = CatalogItemId::User(100);
+
+    // A session creates a temporary item, next to a normal one, and the process
+    // then dies without closing the session.
+    {
+        let mut state = state_builder
+            .clone()
+            .unwrap_build()
+            .await
+            .open(SYSTEM_TIME().into(), &test_bootstrap_args())
+            .await
+            .unwrap();
+        let _ = state
+            .sync_to_current_updates()
+            .await
+            .expect("unable to sync");
+
+        let mut txn = state.transaction().await.unwrap();
+        insert_view(&mut txn, normal_id, SchemaId::User(1), "keep", None);
+        // Temporary items are parented to a sentinel schema id shared by
+        // every session.
+        insert_view(
+            &mut txn,
+            ephemeral_id,
+            SchemaId::User(0),
+            "tt",
+            Some(owner_session),
+        );
+        let _ = txn.get_and_commit_op_updates();
+        let commit_ts = txn.upper();
+        txn.commit(commit_ts).await.unwrap();
+
+        let snapshot = state.snapshot().await.unwrap();
+        assert!(item_ids(&snapshot).contains(&ephemeral_id));
+        Box::new(state).expire().await;
+    }
+
+    // A read-only open leaves it alone. Checked before the writable open below,
+    // which is what removes it.
+    {
+        let mut read_only_state = state_builder
+            .clone()
+            .unwrap_build()
+            .await
+            .open_read_only(&test_bootstrap_args())
+            .await
+            .unwrap();
+        let ids = item_ids(&read_only_state.snapshot().await.unwrap());
+        assert!(
+            ids.contains(&ephemeral_id),
+            "read-only open reclaimed an ephemeral item"
+        );
+        Box::new(read_only_state).expire().await;
+    }
+
+    // Opening with write intent reclaims it, and only it.
+    {
+        let mut state = state_builder
+            .unwrap_build()
+            .await
+            .open(SYSTEM_TIME().into(), &test_bootstrap_args())
+            .await
+            .unwrap();
+        let ids = item_ids(&state.snapshot().await.unwrap());
+        assert!(
+            !ids.contains(&ephemeral_id),
+            "writable open did not reclaim the ephemeral item"
+        );
+        assert!(
+            ids.contains(&normal_id),
+            "writable open reclaimed a non-ephemeral item"
+        );
+        Box::new(state).expire().await;
+    }
+}
+
+#[mz_ore::test(tokio::test)]
+#[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
+async fn test_persist_open_reclaims_late_ephemeral_items() {
+    let persist_client = PersistClient::new_for_tests().await;
+    let state_builder = TestCatalogStateBuilder::new(persist_client);
+    test_open_reclaims_late_ephemeral_items(state_builder).await;
+}
+
+/// During a zero-downtime deploy the outgoing leader keeps serving sessions
+/// while the incoming generation opens the catalog, so a temporary item can
+/// become durable after the incoming opener has synced its initial snapshot
+/// but before its fence lands. The open must reclaim items it did not know
+/// about when the opener was built, since reclamation runs on the state synced
+/// by the fence loop, not on the build-time snapshot.
+async fn test_open_reclaims_late_ephemeral_items(state_builder: TestCatalogStateBuilder) {
+    let owner_session = Uuid::from_u128(1);
+    let ephemeral_id = CatalogItemId::User(200);
+    let deploy_generation = 0;
+
+    // The outgoing leader.
+    let mut state = state_builder
+        .clone()
+        .with_deploy_generation(deploy_generation)
+        .unwrap_build()
+        .await
+        .open(SYSTEM_TIME().into(), &test_bootstrap_args())
+        .await
+        .unwrap();
+    let _ = state
+        .sync_to_current_updates()
+        .await
+        .expect("unable to sync");
+
+    // The incoming generation syncs its snapshot of the catalog here, before
+    // the temporary item below exists. Building does not fence anyone, only
+    // the open below does.
+    let unopened = state_builder
+        .with_deploy_generation(deploy_generation + 1)
+        .unwrap_build()
+        .await;
+
+    // A session on the still unfenced leader creates a temporary item.
+    let mut txn = state.transaction().await.unwrap();
+    insert_view(
+        &mut txn,
+        ephemeral_id,
+        SchemaId::User(0),
+        "tt",
+        Some(owner_session),
+    );
+    let _ = txn.get_and_commit_op_updates();
+    let commit_ts = txn.upper();
+    txn.commit(commit_ts).await.unwrap();
+    assert!(
+        item_ids(&state.snapshot().await.unwrap()).contains(&ephemeral_id),
+        "the temporary item must be durable before the promotion"
+    );
+
+    // Promotion. The open fences the leader and reclaims the item the opener
+    // never saw at build time.
+    let mut new_state = unopened
+        .open(SYSTEM_TIME().into(), &test_bootstrap_args())
+        .await
+        .unwrap();
+    let ids = item_ids(&new_state.snapshot().await.unwrap());
+    assert!(
+        !ids.contains(&ephemeral_id),
+        "open did not reclaim an ephemeral item committed after the opener was built"
+    );
+
+    // The outgoing leader is fenced.
+    let err = state.transaction().await.unwrap_err();
+    assert!(
+        matches!(
+            err,
+            CatalogError::Durable(DurableCatalogError::Fence(
+                FenceError::DeployGeneration { .. }
+            ))
+        ),
+        "unexpected err: {err:?}"
+    );
+    Box::new(new_state).expire().await;
+}
+
+#[mz_ore::test(tokio::test)]
+#[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
+async fn test_persist_fenced_ephemeral_item_write() {
+    let persist_client = PersistClient::new_for_tests().await;
+    let state_builder = TestCatalogStateBuilder::new(persist_client);
+    test_fenced_ephemeral_item_write(state_builder).await;
+}
+
+/// A fenced-out catalog cannot durably create a temporary item. Once the fence
+/// from a newer deploy generation lands, the old owner's in-flight commit is
+/// rejected wholesale, so the item never becomes durable and there is nothing
+/// for the new generation to reclaim.
+async fn test_fenced_ephemeral_item_write(state_builder: TestCatalogStateBuilder) {
+    let owner_session = Uuid::from_u128(1);
+    let ephemeral_id = CatalogItemId::User(200);
+    let deploy_generation = 0;
+
+    let mut state = state_builder
+        .clone()
+        .with_deploy_generation(deploy_generation)
+        .unwrap_build()
+        .await
+        .open(SYSTEM_TIME().into(), &test_bootstrap_args())
+        .await
+        .unwrap();
+    let _ = state
+        .sync_to_current_updates()
+        .await
+        .expect("unable to sync");
+
+    // A session starts creating a temporary item but has not committed yet.
+    let mut txn = state.transaction().await.unwrap();
+    insert_view(
+        &mut txn,
+        ephemeral_id,
+        SchemaId::User(0),
+        "tt",
+        Some(owner_session),
+    );
+    let _ = txn.get_and_commit_op_updates();
+
+    // A newer generation opens before the commit lands.
+    let mut new_state = state_builder
+        .with_deploy_generation(deploy_generation + 1)
+        .unwrap_build()
+        .await
+        .open(SYSTEM_TIME().into(), &test_bootstrap_args())
+        .await
+        .unwrap();
+
+    // The commit is rejected by the fence.
+    let commit_ts = txn.upper();
+    let err = txn.commit(commit_ts).await.unwrap_err();
+    assert!(
+        matches!(
+            err,
+            CatalogError::Durable(DurableCatalogError::Fence(
+                FenceError::DeployGeneration { .. }
+            ))
+        ),
+        "unexpected err: {err:?}"
+    );
+
+    // The item never became durable.
+    let ids = item_ids(&new_state.snapshot().await.unwrap());
+    assert!(
+        !ids.contains(&ephemeral_id),
+        "rejected ephemeral item write became durable"
+    );
+    Box::new(new_state).expire().await;
+}
+
 #[mz_ore::test(tokio::test)]
 #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
 async fn test_persist_open() {