reddb-io-server 1.23.1

RedDB server-side engine: storage, runtime, replication, MCP, AI, and the gRPC/HTTP/RedWire/PG-wire dispatchers. Re-exported by the umbrella `reddb` crate.
Documentation
use super::*;

const SCRUB_COLUMNS: [&str; 15] = [
    "row_kind",
    "zone_kind",
    "physical_identity",
    "collection",
    "expected_checksum",
    "actual_checksum",
    "fault_class",
    "objects_verified",
    "superblock_verified",
    "manifest_verified",
    "wal_verified",
    "page_verified",
    "segment_chunk_verified",
    "bytes_read",
    "duration_ms",
];

impl RedDBRuntime {
    pub(crate) fn execute_scrub_query(
        &self,
        query: &str,
        mode: QueryMode,
        background: bool,
        budget: Option<u64>,
    ) -> RedDBResult<RuntimeQueryResult> {
        let Some(path) = self.inner.db.path() else {
            return Err(RedDBError::InvalidOperation(
                "SCRUB requires a persistent store".to_string(),
            ));
        };
        let max_objects = if background {
            budget.unwrap_or(1).max(1) as usize
        } else {
            usize::MAX
        };
        let start_cursor = if background {
            self.inner.scrub_state.lock().background_cursor
        } else {
            0
        };
        let report = reddb_file::scrub_embedded_store(path, start_cursor, max_objects)
            .map_err(|err| RedDBError::Internal(format!("SCRUB failed: {err}")))?;

        let now_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|duration| duration.as_millis() as u64)
            .unwrap_or(0);
        {
            let mut state = self.inner.scrub_state.lock();
            state.last_run_unix_ms = now_ms;
            state.last_findings_count = report.findings.len() as u64;
            state.verified = report.verified.clone();
            if background {
                state.background_cursor = if report.complete {
                    0
                } else {
                    report.next_cursor
                };
                state.background_status = if report.complete {
                    "complete".to_string()
                } else {
                    "running".to_string()
                };
                state.background_verified_objects = if report.complete {
                    report.total_objects
                } else {
                    report.next_cursor as u64
                };
                state.background_total_objects = report.total_objects;
            }
        }

        let result = scrub_result_rows(&report);
        Ok(RuntimeQueryResult {
            query: query.to_string(),
            mode,
            statement: if background {
                "scrub_background"
            } else {
                "scrub"
            },
            engine: "runtime-scrub",
            result,
            affected_rows: 0,
            statement_type: "select",
            bookmark: None,
            notice: None,
        })
    }

    pub(crate) fn scrub_stats_snapshot(&self) -> ScrubStatsSnapshot {
        self.inner.scrub_state.lock().snapshot()
    }
}

fn scrub_result_rows(report: &reddb_file::StorageScrubReport) -> UnifiedResult {
    let mut result =
        UnifiedResult::with_columns(SCRUB_COLUMNS.iter().map(|s| s.to_string()).collect());
    for finding in &report.findings {
        let mut record = UnifiedRecord::new();
        record.set("row_kind", Value::text("finding"));
        record.set("zone_kind", Value::text(finding.zone_kind.as_str()));
        record.set(
            "physical_identity",
            Value::text(finding.physical_identity.as_str()),
        );
        record.set(
            "collection",
            finding
                .collection
                .as_ref()
                .map(|value| Value::text(value.as_str()))
                .unwrap_or(Value::Null),
        );
        record.set(
            "expected_checksum",
            finding
                .expected_checksum
                .as_ref()
                .map(|value| Value::text(value.as_str()))
                .unwrap_or(Value::Null),
        );
        record.set(
            "actual_checksum",
            finding
                .actual_checksum
                .as_ref()
                .map(|value| Value::text(value.as_str()))
                .unwrap_or(Value::Null),
        );
        record.set(
            "fault_class",
            finding
                .fault_class
                .as_ref()
                .map(|value| Value::text(value.as_str()))
                .unwrap_or(Value::Null),
        );
        record.set(
            "objects_verified",
            Value::UnsignedInteger(report.objects_verified),
        );
        set_verified_counter_fields(&mut record, &report.verified);
        record.set("bytes_read", Value::UnsignedInteger(report.bytes_read));
        record.set("duration_ms", Value::UnsignedInteger(report.duration_ms));
        result.push(record);
    }

    let mut summary = UnifiedRecord::new();
    summary.set("row_kind", Value::text("summary"));
    summary.set("zone_kind", Value::text("summary"));
    summary.set("physical_identity", Value::text("store"));
    summary.set("collection", Value::Null);
    summary.set("expected_checksum", Value::Null);
    summary.set("actual_checksum", Value::Null);
    summary.set(
        "fault_class",
        if report.complete {
            Value::Null
        } else {
            Value::text("in-progress")
        },
    );
    summary.set(
        "objects_verified",
        Value::UnsignedInteger(report.objects_verified),
    );
    set_verified_counter_fields(&mut summary, &report.verified);
    summary.set("bytes_read", Value::UnsignedInteger(report.bytes_read));
    summary.set("duration_ms", Value::UnsignedInteger(report.duration_ms));
    result.push(summary);
    result
}

fn set_verified_counter_fields(
    record: &mut UnifiedRecord,
    verified: &reddb_file::StorageScrubVerifiedCounters,
) {
    record.set(
        "superblock_verified",
        Value::UnsignedInteger(verified.superblock),
    );
    record.set(
        "manifest_verified",
        Value::UnsignedInteger(verified.manifest),
    );
    record.set("wal_verified", Value::UnsignedInteger(verified.wal));
    record.set("page_verified", Value::UnsignedInteger(verified.page));
    record.set(
        "segment_chunk_verified",
        Value::UnsignedInteger(verified.segment_chunk),
    );
}