ahri_tre_pgmeta/

dataset_reservation.rs

use std::sync::Arc;

use ahri_tre_core::{
    CoreError, DatasetAttemptExecutor, DatasetExecutorIdentity, DatasetExecutorLease,
    DatasetOutputAuthority, DatasetOutputKey, DatasetOutputVersion, DatasetWriteAuthority,
};
use uuid::Uuid;

use crate::dataset_admission::ReceiptRow;
use crate::{MetadataExecutor, MetadataTransaction, PgMetaError, PgMetadataRepository};

#[derive(Debug, thiserror::Error)]
pub enum DatasetReservationError {
    #[error("Dataset output is already reserved")]
    Conflict,
    #[error("Dataset output ownership is unavailable")]
    Unavailable(#[source] CoreError),
}

/// Private attempt identity; this is independent of any public operation record.
#[derive(Debug, Clone)]
pub struct RecoverableDatasetAttempt {
    /// Existing opaque Operation join; untracked attempts have none.
    pub operation_id: Option<Uuid>,
    pub attempt_id: Uuid,
    pub owner: DatasetExecutorIdentity,
    pub output: DatasetOutputKey,
    pub version: Option<DatasetOutputVersion>,
    pub creation_intent: bool,
    pub scratch_root_identity: Option<String>,
}

/// An accepted writer holds process exclusion until its last synchronous effect
/// returns. Dropping it leaves the durable reservation for reconciliation.
pub struct PgDatasetReservation<'connection> {
    repository: PgMetadataRepository<'connection>,
    executor: Arc<dyn DatasetExecutorLease>,
    _active: Box<dyn DatasetAttemptExecutor>,
    attempt: RecoverableDatasetAttempt,
    version: Option<DatasetOutputVersion>,
}

impl std::fmt::Debug for PgDatasetReservation<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PgDatasetReservation")
            .field("attempt", &self.attempt)
            .finish_non_exhaustive()
    }
}

/// Exclusive cleanup authority for one known stopped attempt. It has no writer
/// or metadata-admission interface and cannot replay the interrupted operation.
pub struct PgDatasetRecovery<'connection> {
    reservation: PgDatasetReservation<'connection>,
}

impl PgDatasetRecovery<'_> {
    pub fn attempt_id(&self) -> Uuid {
        self.reservation.attempt_id()
    }
    pub fn requires_output_cleanup(&self) -> bool {
        self.reservation.attempt.creation_intent
    }
    pub fn release_after_cleanup(self) -> Result<(), CoreError> {
        self.reservation.release_after_cleanup()
    }
}
impl DatasetOutputAuthority for PgDatasetRecovery<'_> {
    fn authorize_scratch_root(&mut self, identity: &str) -> Result<(), CoreError> {
        if self
            .reservation
            .attempt
            .scratch_root_identity
            .as_deref()
            .is_some_and(|expected| expected != identity)
        {
            return Err(ownership_unavailable());
        }
        self.ensure_current()
    }
    fn attempt_id(&self) -> Uuid {
        self.reservation.attempt_id()
    }
    fn output_version(&self) -> Result<&DatasetOutputVersion, CoreError> {
        self.reservation.output_version()
    }
    fn ensure_current(&mut self) -> Result<(), CoreError> {
        self.reservation.ensure_current()
    }
}

impl<'connection> PgMetadataRepository<'connection> {
    /// Reserves output before version allocation, source staging, or Lake effects.
    /// Conflicting acceptance rolls back without an accepted private attempt.
    pub fn reserve_dataset_output(
        &self,
        executor: Arc<dyn DatasetExecutorLease>,
        output: DatasetOutputKey,
    ) -> Result<PgDatasetReservation<'connection>, DatasetReservationError> {
        self.reserve_dataset_output_with(executor, output, |_, _| Ok(()))
            .map(|(reservation, ())| reservation)
    }

    /// Pending operation/idempotency repositories may participate in acceptance
    /// through this scoped metadata callback. No operation is required by writers.
    pub fn reserve_dataset_output_with<T>(
        &self,
        executor: Arc<dyn DatasetExecutorLease>,
        output: DatasetOutputKey,
        accept: impl FnOnce(
            PgMetadataRepository<'_>,
            &RecoverableDatasetAttempt,
        ) -> Result<T, CoreError>,
    ) -> Result<(PgDatasetReservation<'connection>, T), DatasetReservationError> {
        if output.lake_name.is_empty() {
            return Err(DatasetReservationError::Unavailable(CoreError::Validation(
                "Dataset output identity is invalid".into(),
            )));
        }
        let attempt = RecoverableDatasetAttempt {
            operation_id: None,
            attempt_id: Uuid::new_v4(),
            owner: executor.identity(),
            output,
            version: None,
            creation_intent: false,
            scratch_root_identity: None,
        };
        let active = executor
            .clone()
            .begin_attempt(attempt.attempt_id)
            .map_err(DatasetReservationError::Unavailable)?;
        let accepted = self
            .with_dataset_acceptance(|repository| {
                let reserved = repository.with_session_connection(
                    "reserve Dataset output",
                    |c| reserve(c, &attempt),
                    |c| reserve(c, &attempt),
                )?;
                if !reserved {
                    return Err(CoreError::Conflict(
                        "Dataset output is already reserved".into(),
                    ));
                }
                accept(repository, &attempt)
            })
            .map_err(|error| match error {
                CoreError::Conflict(_) => DatasetReservationError::Conflict,
                error => DatasetReservationError::Unavailable(error),
            })?;
        Ok((
            PgDatasetReservation {
                repository: self.clone(),
                executor,
                _active: active,
                attempt,
                version: None,
            },
            accepted,
        ))
    }

    /// Lists only attempts whose executor is excluded by the retained local
    /// coordinator. Metadata RLS limits ordinary actors to their own evidence;
    /// configured maintenance connections may see all actors' attempts.
    pub fn recoverable_dataset_attempts(
        &self,
        executor: Arc<dyn DatasetExecutorLease>,
    ) -> Result<Vec<RecoverableDatasetAttempt>, CoreError> {
        let attempts =
            self.with_session_connection("inspect Dataset attempts", read_attempts, read_attempts)?;
        Ok(attempts
            .into_iter()
            .filter(|attempt| executor.can_recover(attempt.owner, attempt.attempt_id))
            .collect())
    }

    /// Claims a known stopped attempt before any physical reconciliation. The
    /// reservation row orders this claim against an in-flight metadata commit.
    pub fn claim_dataset_recovery(
        &self,
        executor: Arc<dyn DatasetExecutorLease>,
        attempt_id: Uuid,
    ) -> Result<PgDatasetRecovery<'connection>, CoreError> {
        let attempt = self
            .recoverable_dataset_attempts(executor.clone())?
            .into_iter()
            .find(|attempt| attempt.attempt_id == attempt_id)
            .ok_or_else(ownership_unavailable)?;
        let active = executor.clone().begin_attempt(attempt_id)?;
        let owner = executor.identity();
        let claimed = self.with_session_connection(
            "claim Dataset cleanup",
            |connection| claim_recovery(connection, &attempt, owner),
            |connection| claim_recovery(connection, &attempt, owner),
        )?;
        let version = claimed.version.clone();
        Ok(PgDatasetRecovery {
            reservation: PgDatasetReservation {
                repository: self.clone(),
                executor,
                _active: active,
                attempt: claimed,
                version,
            },
        })
    }
}

impl PgDatasetReservation<'_> {
    pub fn attempt_id(&self) -> Uuid {
        self.attempt.attempt_id
    }

    pub fn requires_output_cleanup(&self) -> bool {
        self.attempt.creation_intent
    }

    /// Binds version coordinates chosen after reservation, before any Lake effect.
    pub fn bind_version(
        &mut self,
        asset: &ahri_tre_types::AssetRecord,
        version: &ahri_tre_types::AssetVersionRecord,
    ) -> Result<(), CoreError> {
        if self.version.is_some()
            || asset.study_id != self.attempt.output.study_id
            || asset.asset_id != version.asset_id
        {
            return Err(ownership_unavailable());
        }
        self.ensure_current()?;
        self.repository.with_session_connection(
            "bind Dataset output version",
            |connection| bind_version(connection, &self.attempt, asset, version),
            |connection| bind_version(connection, &self.attempt, asset, version),
        )?;
        self.version = Some(DatasetOutputVersion {
            output: self.attempt.output.clone(),
            version_id: version.version_id,
            major: version.major,
            minor: version.minor,
            patch: version.patch,
        });
        Ok(())
    }

    pub(super) fn consume(
        &self,
        connection: &mut impl MetadataExecutor,
    ) -> Result<(), PgMetaError> {
        let version = self.version.as_ref().ok_or_else(ownership_error)?;
        let changed = connection.execute_command("admit owned Dataset output", "
            UPDATE public.dataset_output_attempts a SET admitted = TRUE
            WHERE a.attempt_id = $1 AND a.version_id = $2 AND a.coordinator_id = $3 AND a.generation_id = $4
              AND NOT a.admitted AND EXISTS (SELECT 1 FROM public.dataset_output_reservations r WHERE r.attempt_id = a.attempt_id)
              AND EXISTS (SELECT 1 FROM public.dataset_executor_generation g WHERE g.coordinator_id = a.coordinator_id AND g.generation_id = a.generation_id)",
            &[&self.attempt.attempt_id, &version.version_id.0, &self.attempt.owner.coordinator_id, &self.attempt.owner.generation_id])?;
        if changed != 1 {
            return Err(ownership_error());
        }
        let consumed = connection.execute_command(
            "consume Dataset output reservation",
            "DELETE FROM public.dataset_output_reservations WHERE attempt_id = $1",
            &[&self.attempt.attempt_id],
        )?;
        if consumed != 1 {
            return Err(ownership_error());
        }
        Ok(())
    }

    /// Called only after this attempt's source and physical output cleanup have
    /// completed. Consuming the writer excludes subsequent effects through it.
    pub fn release_after_cleanup(self) -> Result<(), CoreError> {
        if self.executor.identity() != self.attempt.owner {
            return Err(ownership_unavailable());
        }
        self.repository.with_session_connection(
            "release Dataset output",
            |connection| release(connection, &self.attempt),
            |connection| release(connection, &self.attempt),
        )
    }
}

impl DatasetOutputAuthority for PgDatasetReservation<'_> {
    fn authorize_scratch_root(&mut self, identity: &str) -> Result<(), CoreError> {
        self.ensure_current()?;
        if identity.len() != 64 {
            return Err(ownership_unavailable());
        }
        self.repository.with_session_connection(
            "bind Dataset source root",
            |c| bind_scratch(c, self.attempt.attempt_id, identity),
            |c| bind_scratch(c, self.attempt.attempt_id, identity),
        )?;
        self.attempt.scratch_root_identity = Some(identity.to_string());
        Ok(())
    }
    fn attempt_id(&self) -> Uuid {
        self.attempt.attempt_id
    }
    fn output_version(&self) -> Result<&DatasetOutputVersion, CoreError> {
        self.version.as_ref().ok_or_else(ownership_unavailable)
    }

    fn ensure_current(&mut self) -> Result<(), CoreError> {
        self.repository.with_session_connection(
            "check Dataset output ownership",
            |connection| ensure_current(connection, &self.attempt),
            |connection| ensure_current(connection, &self.attempt),
        )
    }
}

impl DatasetWriteAuthority for PgDatasetReservation<'_> {
    fn record_creation_intent(&mut self) -> Result<(), CoreError> {
        self.output_version()?;
        self.ensure_current()?;
        self.repository.with_session_connection(
            "record Dataset creation intent",
            |connection| record_creation_intent(connection, &self.attempt),
            |connection| record_creation_intent(connection, &self.attempt),
        )?;
        self.attempt.creation_intent = true;
        Ok(())
    }
}

fn bind_version(
    connection: &mut impl MetadataExecutor,
    attempt: &RecoverableDatasetAttempt,
    asset: &ahri_tre_types::AssetRecord,
    version: &ahri_tre_types::AssetVersionRecord,
) -> Result<(), PgMetaError> {
    let updated = connection.execute_command("bind Dataset output version", "
        UPDATE public.dataset_output_attempts SET version_id = $2, dataset_name = $3, major = $4, minor = $5, patch = $6
        WHERE attempt_id = $1 AND version_id IS NULL AND NOT admitted", &[&attempt.attempt_id, &version.version_id.0, &asset.name.as_str(), &version.major, &version.minor, &version.patch])?;
    if updated != 1 {
        return Err(ownership_error());
    }
    Ok(())
}

fn ensure_current(
    connection: &mut impl MetadataExecutor,
    attempt: &RecoverableDatasetAttempt,
) -> Result<(), PgMetaError> {
    let current = connection.query_optional("check Dataset output ownership", "
        SELECT a.attempt_id FROM public.dataset_output_attempts a
        JOIN public.dataset_output_reservations r USING (attempt_id)
        JOIN public.dataset_executor_generation g ON g.coordinator_id = a.coordinator_id AND g.generation_id = a.generation_id
        WHERE a.attempt_id = $1 AND a.coordinator_id = $2 AND a.generation_id = $3 AND NOT a.admitted
          AND NOT EXISTS (SELECT 1 FROM public.dataset_admission_receipts d WHERE d.version_id = a.version_id AND d.admitted)",
        &[&attempt.attempt_id, &attempt.owner.coordinator_id, &attempt.owner.generation_id])?;
    if current.is_none() {
        return Err(ownership_error());
    }
    Ok(())
}

fn record_creation_intent(
    connection: &mut impl MetadataExecutor,
    attempt: &RecoverableDatasetAttempt,
) -> Result<(), PgMetaError> {
    let changed = connection.execute_command("record Dataset creation intent", "UPDATE public.dataset_output_attempts SET creation_intent = TRUE WHERE attempt_id = $1 AND version_id IS NOT NULL AND NOT admitted", &[&attempt.attempt_id])?;
    if changed != 1 {
        return Err(ownership_error());
    }
    Ok(())
}

fn establish_generation<R>(
    transaction: &mut dyn MetadataTransaction<Row = R>,
    owner: DatasetExecutorIdentity,
) -> Result<(), PgMetaError> {
    transaction.execute_command("initialize Dataset executor", "
            INSERT INTO public.dataset_executor_generation (singleton, coordinator_id, generation_id)
            VALUES (TRUE, $1, $2) ON CONFLICT (singleton) DO NOTHING", &[&owner.coordinator_id, &owner.generation_id])?;
    transaction.execute_command(
        "replace excluded Dataset executor",
        "
            UPDATE public.dataset_executor_generation SET generation_id = $2
            WHERE coordinator_id = $1 AND generation_id <> $2",
        &[&owner.coordinator_id, &owner.generation_id],
    )?;
    if transaction.query_optional("verify Dataset executor", "SELECT singleton FROM public.dataset_executor_generation WHERE coordinator_id = $1 AND generation_id = $2", &[&owner.coordinator_id, &owner.generation_id])?.is_none() {
            return Err(ownership_error());
        }
    Ok(())
}

fn reserve<E: MetadataExecutor>(
    connection: &mut E,
    attempt: &RecoverableDatasetAttempt,
) -> Result<bool, PgMetaError> {
    connection.with_transaction("reserve Dataset output", |transaction| {
        // This transaction lock closes the race with lifecycle guard acquisition.
        // The durable reservation then excludes deletion until terminal cleanup.
        if transaction.query_optional("check Study lifecycle exclusion", "SELECT 1 WHERE pg_try_advisory_xact_lock_shared(hashtextextended(($1::uuid)::text, 606))", &[&attempt.output.study_id.0])?.is_none() {
            return Ok(false);
        }
        establish_generation(transaction, attempt.owner)?;
        let reserved = transaction.execute_command("reserve Dataset output", "
            INSERT INTO public.dataset_output_reservations (study_id, lake_name, attempt_id)
            VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
            &[&attempt.output.study_id.0, &attempt.output.lake_name, &attempt.attempt_id])?;
        if reserved == 0 { return Ok(false); }
        transaction.execute_command("record private Dataset attempt", "
            INSERT INTO public.dataset_output_attempts (attempt_id, coordinator_id, generation_id, study_id, lake_name)
            VALUES ($1, $2, $3, $4, $5)", &[&attempt.attempt_id, &attempt.owner.coordinator_id, &attempt.owner.generation_id, &attempt.output.study_id.0, &attempt.output.lake_name])?;
        Ok(true)
    })
}

fn read_attempts<E: MetadataExecutor>(
    connection: &mut E,
) -> Result<Vec<RecoverableDatasetAttempt>, PgMetaError>
where
    E::Row: ReceiptRow,
{
    let rows = connection.query_many("inspect Dataset attempts", "
        SELECT a.attempt_id::text, a.coordinator_id::text, a.generation_id::text, a.study_id::text, a.lake_name,
            coalesce(a.version_id::text, ''), coalesce(a.major::text, ''), coalesce(a.minor::text, ''), coalesce(a.patch::text, ''), a.creation_intent::text, coalesce(a.scratch_root_identity, ''), coalesce(o.operation_id::text, '')
        FROM public.dataset_output_attempts a JOIN public.dataset_output_reservations r USING (attempt_id)
        LEFT JOIN public.operations o USING (attempt_id)
        ORDER BY a.created_at, a.attempt_id", &[])?;
    rows.into_iter()
        .map(|row| {
            let mut attempt = decode_attempt(&row)?;
            attempt.operation_id = if row.text(11)?.is_empty() {
                None
            } else {
                Some(row.text(11)?.parse().map_err(|_| ownership_error())?)
            };
            Ok(attempt)
        })
        .collect()
}

fn decode_attempt(row: &impl ReceiptRow) -> Result<RecoverableDatasetAttempt, PgMetaError> {
    let output = DatasetOutputKey {
        study_id: ahri_tre_types::StudyId(row.text(3)?.parse().map_err(|_| ownership_error())?),
        lake_name: row.text(4)?.into(),
    };
    let version = if row.text(5)?.is_empty() {
        None
    } else {
        Some(DatasetOutputVersion {
            output: output.clone(),
            version_id: ahri_tre_types::VersionId(
                row.text(5)?.parse().map_err(|_| ownership_error())?,
            ),
            major: row.text(6)?.parse().map_err(|_| ownership_error())?,
            minor: row.text(7)?.parse().map_err(|_| ownership_error())?,
            patch: row.text(8)?.parse().map_err(|_| ownership_error())?,
        })
    };
    Ok(RecoverableDatasetAttempt {
        operation_id: None,
        attempt_id: row.text(0)?.parse().map_err(|_| ownership_error())?,
        owner: DatasetExecutorIdentity {
            coordinator_id: row.text(1)?.parse().map_err(|_| ownership_error())?,
            generation_id: row.text(2)?.parse().map_err(|_| ownership_error())?,
        },
        output,
        version,
        creation_intent: row.text(9)? == "true",
        scratch_root_identity: (!row.text(10)?.is_empty())
            .then(|| row.text(10).unwrap().to_string()),
    })
}

fn claim_recovery<E: MetadataExecutor>(
    connection: &mut E,
    prior: &RecoverableDatasetAttempt,
    owner: DatasetExecutorIdentity,
) -> Result<RecoverableDatasetAttempt, PgMetaError>
where
    E::Row: ReceiptRow,
{
    connection.with_transaction("claim Dataset cleanup", |transaction| {
        // Use the same generation-before-reservation lock order as acceptance.
        establish_generation(transaction, owner)?;
        // Lock before reading outcome; a final admission that already consumed ownership wins.
        if transaction.query_optional("lock interrupted Dataset reservation", "SELECT attempt_id FROM public.dataset_output_reservations WHERE attempt_id = $1 FOR UPDATE", &[&prior.attempt_id])?.is_none() { return Err(ownership_error()); }
        let row = transaction.query_optional("inspect interrupted Dataset attempt", "
            SELECT a.attempt_id::text, a.coordinator_id::text, a.generation_id::text, a.study_id::text, a.lake_name,
                coalesce(a.version_id::text, ''), coalesce(a.major::text, ''), coalesce(a.minor::text, ''), coalesce(a.patch::text, ''), a.creation_intent::text, coalesce(a.scratch_root_identity, '')
            FROM public.dataset_output_attempts a WHERE a.attempt_id = $1 AND a.coordinator_id = $2 AND a.generation_id = $3
                AND NOT a.admitted AND NOT EXISTS (SELECT 1 FROM public.dataset_admission_receipts d WHERE d.version_id = a.version_id AND d.admitted)",
            &[&prior.attempt_id, &prior.owner.coordinator_id, &prior.owner.generation_id])?.ok_or_else(ownership_error)?;
        let mut claimed = decode_attempt(&row)?;
        transaction.execute_command("transfer stopped Dataset attempt", "UPDATE public.dataset_output_attempts SET generation_id = $2 WHERE attempt_id = $1", &[&prior.attempt_id, &owner.generation_id])?;
        claimed.owner = owner;
        Ok(claimed)
    })
}

fn release<E: MetadataExecutor>(
    connection: &mut E,
    attempt: &RecoverableDatasetAttempt,
) -> Result<(), PgMetaError> {
    connection.with_transaction("release Dataset output", |transaction| {
        // Row locking orders reconciliation after any in-flight final admission.
        transaction.query_optional("lock Dataset reservation", "SELECT attempt_id FROM public.dataset_output_reservations WHERE attempt_id = $1 FOR UPDATE", &[&attempt.attempt_id])?;
        let removed = transaction.execute_command("release Dataset output", "
            DELETE FROM public.dataset_output_reservations r USING public.dataset_output_attempts a
            WHERE r.attempt_id = a.attempt_id AND a.attempt_id = $1
              AND a.coordinator_id = $2 AND a.generation_id = $3 AND NOT a.admitted
              AND NOT EXISTS (SELECT 1 FROM public.dataset_admission_receipts d WHERE d.version_id = a.version_id AND d.admitted)",
            &[&attempt.attempt_id, &attempt.owner.coordinator_id, &attempt.owner.generation_id])?;
        if removed != 1 { return Err(ownership_error()); }
        transaction.execute_command("remove cleaned admission evidence", "DELETE FROM public.dataset_admission_receipts d USING public.dataset_output_attempts a WHERE a.attempt_id = $1 AND d.version_id = a.version_id AND NOT d.admitted", &[&attempt.attempt_id])?;
        transaction.execute_command("remove cleaned private Dataset attempt", "DELETE FROM public.dataset_output_attempts a WHERE attempt_id = $1 AND NOT EXISTS (SELECT 1 FROM public.operations o WHERE o.attempt_id=a.attempt_id AND o.record#>>'{detail,summary,status}' NOT IN ('completed','failed','cancelled'))", &[&attempt.attempt_id])?;
        Ok(())
    })
}

fn ownership_unavailable() -> CoreError {
    CoreError::Infrastructure("Dataset output ownership could not be established".into())
}
fn ownership_error() -> PgMetaError {
    PgMetaError::Decode {
        field: "dataset_output",
        value: String::new(),
        message: "Dataset output ownership could not be established".into(),
    }
}

/// Configured maintenance authority exposes only known-attempt reconciliation.
/// The underlying administrator connection never escapes this adapter boundary.
pub struct PgDatasetMaintenance {
    pub(crate) repository: PgMetadataRepository<'static>,
}
impl PgDatasetMaintenance {
    pub(crate) fn new(mut connection: crate::PgMetadataConnection) -> Result<Self, PgMetaError> {
        let row = connection.client().query_one("SELECT count(*) = 4 AND coalesce(bool_and(r.rolsuper OR r.rolbypassrls OR (NOT c.relforcerowsecurity AND pg_has_role(c.relowner, 'USAGE'))), false) FROM pg_class c CROSS JOIN pg_roles r WHERE r.rolname = CURRENT_USER AND c.oid IN (to_regclass('public.dataset_output_attempts'), to_regclass('public.dataset_output_reservations'), to_regclass('public.dataset_admission_receipts'), to_regclass('public.dataset_executor_generation'))", &[]).map_err(PgMetaError::Query)?;
        if !row.get::<_, bool>(0) {
            return Err(ownership_error());
        }
        Ok(Self {
            repository: PgMetadataRepository::new(connection),
        })
    }
    pub fn recoverable_attempts(
        &self,
        executor: Arc<dyn DatasetExecutorLease>,
    ) -> Result<Vec<RecoverableDatasetAttempt>, CoreError> {
        self.repository.recoverable_dataset_attempts(executor)
    }
    pub fn claim(
        &self,
        executor: Arc<dyn DatasetExecutorLease>,
        attempt_id: Uuid,
    ) -> Result<PgDatasetRecovery<'static>, CoreError> {
        self.repository.claim_dataset_recovery(executor, attempt_id)
    }
    pub fn has_unresolved_attempts(&self) -> Result<bool, CoreError> {
        self.repository.with_session_connection(
            "inspect unresolved Dataset outputs",
            |c| read_attempts(c).map(|v| !v.is_empty()),
            |c| read_attempts(c).map(|v| !v.is_empty()),
        )
    }
}

/// Administrative Lake replacement has no materialization/admission methods.
pub struct PgDatasetReset {
    repository: PgMetadataRepository<'static>,
    _exclusion: Box<dyn DatasetAttemptExecutor>,
}
impl PgDatasetMaintenance {
    pub fn authorize_namespace_reset(
        &self,
        executor: Arc<dyn DatasetExecutorLease>,
    ) -> Result<PgDatasetReset, CoreError> {
        ahri_tre_observability::LifecycleSpan::run(
            ahri_tre_observability::Phase::Maintenance,
            None,
            ahri_tre_observability::FailureCategory::Admission,
            || {
                let exclusion = executor.clone().begin_maintenance()?;
                self.repository.with_session_connection(
                    "exclude Dataset writers for reset",
                    |c| {
                        c.with_transaction("establish Dataset reset executor", |t| {
                            establish_generation(t, executor.identity())
                        })
                    },
                    |c| {
                        c.with_transaction("establish Dataset reset executor", |t| {
                            establish_generation(t, executor.identity())
                        })
                    },
                )?;
                let mut reset = PgDatasetReset {
                    repository: self.repository.clone(),
                    _exclusion: exclusion,
                };
                ahri_tre_core::DatasetMaintenanceAuthority::ensure_exclusive(&mut reset)?;
                Ok(reset)
            },
        )
    }
}
impl ahri_tre_core::DatasetMaintenanceAuthority for PgDatasetReset {
    fn ensure_exclusive(&mut self) -> Result<(), CoreError> {
        let unresolved = self.repository.with_session_connection(
            "verify exclusive Dataset reset",
            |c| read_attempts(c).map(|v| !v.is_empty()),
            |c| read_attempts(c).map(|v| !v.is_empty()),
        )?;
        if unresolved {
            Err(CoreError::Conflict(
                "Dataset output remains reserved".into(),
            ))
        } else {
            Ok(())
        }
    }
}

fn bind_scratch(
    connection: &mut impl MetadataExecutor,
    attempt: Uuid,
    identity: &str,
) -> Result<(), PgMetaError> {
    let changed = connection.execute_command("bind Dataset source root", "UPDATE public.dataset_output_attempts SET scratch_root_identity = $2 WHERE attempt_id = $1 AND NOT admitted AND (scratch_root_identity IS NULL OR scratch_root_identity = $2)", &[&attempt, &identity])?;
    if changed != 1 {
        return Err(ownership_error());
    }
    Ok(())
}