diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 38cb36452..cbd2ed285 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -512,7 +512,7 @@ jobs: extension-artifacts-native-linux: name: Native Extensions (Linux) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.extension_artifacts_native_matrix_linux).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.extension_artifacts_native_matrix_linux).include[0] != null }} uses: ./.github/workflows/extension-artifacts-native.yml with: cache-save-if: ${{ needs.affected.outputs.cache_save_if == 'true' }} @@ -522,7 +522,7 @@ jobs: extension-artifacts-native-android-static: name: Native Extensions (Android Compilation) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.extension_artifacts_native_matrix_android).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.extension_artifacts_native_matrix_android).include[0] != null }} uses: ./.github/workflows/extension-artifacts-native.yml with: phase: android-static @@ -544,7 +544,7 @@ jobs: extension-artifacts-native-ios: name: Native Extensions (iOS) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.extension_artifacts_native_matrix_ios).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.extension_artifacts_native_matrix_ios).include[0] != null }} uses: ./.github/workflows/extension-artifacts-native.yml with: cache-save-if: ${{ needs.affected.outputs.cache_save_if == 'true' }} @@ -554,7 +554,7 @@ jobs: extension-artifacts-native-other: name: Native Extensions (Desktop) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.extension_artifacts_native_matrix_other).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.extension_artifacts_native_matrix_other).include[0] != null }} uses: ./.github/workflows/extension-artifacts-native.yml with: cache-save-if: ${{ needs.affected.outputs.cache_save_if == 'true' }} @@ -732,7 +732,7 @@ jobs: liboliphaunt-native-desktop-linux: name: Native Runtime (Linux) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.liboliphaunt_native_desktop_runtime_matrix_linux).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.liboliphaunt_native_desktop_runtime_matrix_linux).include[0] != null }} uses: ./.github/workflows/liboliphaunt-native-desktop.yml with: cache-save-if: ${{ needs.affected.outputs.cache_save_if == 'true' }} @@ -742,7 +742,7 @@ jobs: liboliphaunt-native-desktop-other: name: Native Runtime (Desktop) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.liboliphaunt_native_desktop_runtime_matrix_other).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.liboliphaunt_native_desktop_runtime_matrix_other).include[0] != null }} uses: ./.github/workflows/liboliphaunt-native-desktop.yml with: cache-save-if: ${{ needs.affected.outputs.cache_save_if == 'true' }} @@ -770,7 +770,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-native-android') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-native-android') }} strategy: fail-fast: false matrix: ${{ fromJson(needs.affected.outputs.liboliphaunt_native_android_runtime_matrix) }} @@ -857,7 +857,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-native-ios') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-native-ios') }} strategy: fail-fast: false matrix: ${{ fromJson(needs.affected.outputs.liboliphaunt_native_ios_runtime_matrix) }} @@ -1165,7 +1165,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'rust-sdk-package') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'rust-sdk-package') }} runs-on: ubuntu-24.04 timeout-minutes: 90 steps: @@ -1249,7 +1249,7 @@ jobs: broker-runtime-linux: name: Broker Runtime (Linux) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.broker_runtime_matrix_linux).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.broker_runtime_matrix_linux).include[0] != null }} uses: ./.github/workflows/broker-runtime.yml with: cache-save-if: ${{ needs.affected.outputs.cache_save_if == 'true' }} @@ -1259,7 +1259,7 @@ jobs: broker-runtime-other: name: Broker Runtime (Desktop) needs: [affected, checks, tests] - if: ${{ fromJson(needs.affected.outputs.broker_runtime_matrix_other).include[0] != null }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && fromJson(needs.affected.outputs.broker_runtime_matrix_other).include[0] != null }} uses: ./.github/workflows/broker-runtime.yml with: cache-save-if: ${{ needs.affected.outputs.cache_save_if == 'true' }} @@ -1287,7 +1287,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'node-direct') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'node-direct') }} strategy: fail-fast: false matrix: ${{ fromJson(needs.affected.outputs.node_direct_runtime_matrix) }} @@ -1410,7 +1410,7 @@ jobs: wasix-napi: name: Builds / WASIX Node-API needs: [affected, wasix-host-linux, wasix-host-other] - if: ${{ !cancelled() && !failure() && contains(fromJson(needs.affected.outputs.jobs), 'wasix-napi') }} + if: ${{ !cancelled() && !failure() && needs.affected.result == 'success' && (fromJson(needs.affected.outputs.liboliphaunt_wasix_aot_runtime_matrix_linux).include[0] == null || needs.wasix-host-linux.result == 'success') && (fromJson(needs.affected.outputs.liboliphaunt_wasix_aot_runtime_matrix_other).include[0] == null || needs.wasix-host-other.result == 'success') && contains(fromJson(needs.affected.outputs.jobs), 'wasix-napi') }} runs-on: ubuntu-24.04 timeout-minutes: 5 steps: @@ -1466,7 +1466,7 @@ jobs: swift-bindings: name: Builds / Swift Native Bindings needs: [affected, checks, tests] - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'swift-bindings') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'swift-bindings') }} runs-on: macos-26 timeout-minutes: 30 steps: @@ -1550,7 +1550,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'kotlin-sdk-package') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'kotlin-sdk-package') }} runs-on: ubuntu-24.04 timeout-minutes: 90 steps: @@ -1693,7 +1693,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'js-sdk-package') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'js-sdk-package') }} runs-on: ubuntu-24.04 timeout-minutes: 60 steps: @@ -1867,7 +1867,7 @@ jobs: wasix-ts-package: name: Packages / WASIX TypeScript SDK needs: [affected, checks, tests] - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'wasix-ts-package') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'wasix-ts-package') }} runs-on: ubuntu-24.04 timeout-minutes: 60 steps: @@ -2027,7 +2027,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'wasix-rust-package') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'wasix-rust-package') }} runs-on: ubuntu-24.04 timeout-minutes: 90 steps: @@ -2073,7 +2073,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-wasix-runtime') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-wasix-runtime') }} runs-on: ubuntu-24.04 timeout-minutes: 360 env: @@ -2244,7 +2244,7 @@ jobs: liboliphaunt-wasix-aot: name: Builds / WASIX AOT needs: [affected, wasix-host-linux, wasix-host-other] - if: ${{ !cancelled() && !failure() && contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-wasix-aot') }} + if: ${{ !cancelled() && !failure() && needs.affected.result == 'success' && (fromJson(needs.affected.outputs.liboliphaunt_wasix_aot_runtime_matrix_linux).include[0] == null || needs.wasix-host-linux.result == 'success') && (fromJson(needs.affected.outputs.liboliphaunt_wasix_aot_runtime_matrix_other).include[0] == null || needs.wasix-host-other.result == 'success') && contains(fromJson(needs.affected.outputs.jobs), 'liboliphaunt-wasix-aot') }} runs-on: ubuntu-24.04 timeout-minutes: 5 steps: @@ -2333,7 +2333,7 @@ jobs: - affected - checks - tests - if: ${{ contains(fromJson(needs.affected.outputs.jobs), 'wasix-postmaster') }} + if: ${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && contains(fromJson(needs.affected.outputs.jobs), 'wasix-postmaster') }} runs-on: ubuntu-24.04 timeout-minutes: 240 steps: diff --git a/src/wasix/sdks/rust/src/oliphaunt/base.rs b/src/wasix/sdks/rust/src/oliphaunt/base.rs index 36f725814..623fcd44c 100644 --- a/src/wasix/sdks/rust/src/oliphaunt/base.rs +++ b/src/wasix/sdks/rust/src/oliphaunt/base.rs @@ -27,8 +27,8 @@ use crate::oliphaunt::database_root_descriptor::{ #[cfg(feature = "extensions")] use crate::oliphaunt::extensions::Extension; use crate::oliphaunt::storage::{ - DatabaseStorage, PgDataStorage, StorageRoot, vfs_create_dir_all, vfs_file_exists, vfs_read, - vfs_remove_file_if_exists, vfs_write, + DatabaseStorage, PgDataStorage, StorageRoot, for_each_publication_file, vfs_create_dir_all, + vfs_file_exists, vfs_read, vfs_remove_file_if_exists, vfs_write, }; use crate::{StorageCommitState, StorageErrorCode, StorageErrorPhase}; use tempfile::TempDir; @@ -951,21 +951,47 @@ fn cluster_seed_publication_staging(pgdata: &Path) -> Result { } fn sync_publication_tree(path: &Path) -> Result<()> { + let mut files = Vec::new(); + let mut directories = Vec::new(); + collect_publication_entries(path, &mut files, &mut directories)?; + + // Start writing the whole private staging tree before waiting on each file. + // Every file and directory still receives its checked durability barrier. + for file in &files { + if let Ok(file) = fs::File::open(file) { + pre_sync_publication_file(&file); + } + } + for_each_publication_file(&files, |file| { + sync_publication_file(file) + .with_context(|| format!("sync publication file {}", file.display())) + })?; + for directory in directories { + sync_directory(&directory) + .with_context(|| format!("sync publication directory {}", directory.display()))?; + } + Ok(()) +} + +fn collect_publication_entries( + path: &Path, + files: &mut Vec, + directories: &mut Vec, +) -> Result<()> { let metadata = fs::symlink_metadata(path) .with_context(|| format!("inspect publication entry {}", path.display()))?; ensure!( !metadata.file_type().is_symlink(), - "cluster seed publication contains a symbolic link: {}", + "publication tree contains a symbolic link: {}", path.display() ); if metadata.is_file() { - sync_publication_file(path) - .with_context(|| format!("sync publication file {}", path.display()))?; + files.push(path.to_owned()); return Ok(()); } ensure!( metadata.is_dir(), - "cluster seed publication contains a special file: {}", + "publication tree contains a special file: {}", path.display() ); let mut entries = fs::read_dir(path) @@ -973,11 +999,36 @@ fn sync_publication_tree(path: &Path) -> Result<()> { .collect::>>()?; entries.sort_by_key(|entry| entry.file_name()); for entry in entries { - sync_publication_tree(&entry.path())?; + collect_publication_entries(&entry.path(), files, directories)?; } - sync_directory(path).with_context(|| format!("sync publication directory {}", path.display())) + directories.push(path.to_owned()); + Ok(()) +} + +#[cfg(all(target_os = "linux", target_pointer_width = "64"))] +#[allow(unsafe_code)] +pub(super) fn pre_sync_publication_file(file: &fs::File) { + use std::os::fd::AsRawFd; + unsafe extern "C" { + fn sync_file_range( + fd: std::ffi::c_int, + offset: i64, + len: i64, + flags: u32, + ) -> std::ffi::c_int; + } + const SYNC_FILE_RANGE_WRITE: u32 = 2; + + // SAFETY: the borrowed descriptor stays open for this call; zero offset + // and length cover the regular file. This optional Linux writeback hint + // is never a durability barrier. Unsupported hints fall back to the + // checked sync_all calls below, including their metadata/error handling. + let _ = unsafe { sync_file_range(file.as_raw_fd(), 0, 0, SYNC_FILE_RANGE_WRITE) }; } +#[cfg(not(all(target_os = "linux", target_pointer_width = "64")))] +pub(super) fn pre_sync_publication_file(_file: &fs::File) {} + #[cfg(windows)] fn sync_publication_file(path: &Path) -> std::io::Result<()> { fs::OpenOptions::new() @@ -1805,8 +1856,8 @@ fn prepare_database_root( seed: Option<&ClusterSeed>, icu_data: Option<&IcuData>, ) -> Result { - let mut runtime_layout = prepare_runtime_layout(&paths, profile, icu_data)?; - prepare_pgdata(&paths, initialize, profile, &mut runtime_layout, seed)?; + let runtime_layout = prepare_runtime_layout(&paths, profile, icu_data)?; + prepare_pgdata(&paths, initialize, profile, &runtime_layout, seed)?; Ok(InstallOutcome { runtime_layout, pgdata_storage: PgDataStorage::host_directory(paths.pgdata), @@ -1817,7 +1868,7 @@ fn prepare_pgdata( paths: &OliphauntPaths, initialize: bool, profile: CatalogProfile, - runtime_layout: &mut RuntimeLayout, + runtime_layout: &RuntimeLayout, seed: Option<&ClusterSeed>, ) -> Result<()> { ensure!( @@ -1837,10 +1888,48 @@ fn prepare_pgdata( if try_install_cluster_seed(paths, &runtime_layout.module_path(), profile, seed)? { return Ok(()); } - PostgresMod::run_split_initdb( - runtime_layout, - &PgDataStorage::host_directory(paths.pgdata.clone()), - )?; + // Initialization is one unpublished transaction. Running it on private + // memory avoids thousands of host syncs on intermediate catalog files; + // the finished tree is durably published before any backend can open it. + let pgdata_storage = PgDataStorage::memory(); + let mut initdb_layout = runtime_layout.clone(); + if initdb_layout.uses_shared_overlay() { + // Scratch files belong to the private initialization transaction too. + // Keep the immutable runtime and explicit ICU selection shared. + initdb_layout.mutable_root = StorageRoot::memory(); + initdb_layout.mutable_root.create_runtime_dirs()?; + } + PostgresMod::run_split_initdb(&initdb_layout, &pgdata_storage)?; + let filesystem = pgdata_storage + .memory_filesystem() + .expect("memory storage has a virtual filesystem"); + ensure!( + virtual_cluster_is_complete(filesystem.as_ref()), + "split WASIX initdb did not create a complete PGDATA cluster" + ); + remove_virtual_runtime_state(filesystem.as_ref())?; + ensure_virtual_pgdata_matches_runtime(filesystem.as_ref(), profile)?; + let root = paths + .pgdata + .parent() + .context("PGDATA has no managed-root parent")?; + let staging = tempfile::Builder::new() + .prefix(".pgdata-initdb-") + .tempdir_in(root) + .context("create private initdb publication staging")?; + let publication = (|| -> Result<()> { + super::data_dir::materialize_initialized_pgdata_into(filesystem.as_ref(), staging.path())?; + super::data_dir::apply_private_permissions(staging.path(), 0o700)?; + promote_synced_directory(staging.path(), &paths.pgdata, root, "initialized cluster") + })(); + if let Err(mut error) = publication { + if let Err(cleanup) = staging.close() + && cleanup.kind() != std::io::ErrorKind::NotFound + { + error = error.context(format!("remove failed initdb staging: {cleanup}")); + } + return Err(error); + } ensure!( cluster_is_complete(paths), "split WASIX initdb finished but did not create a complete PGDATA cluster at {}", @@ -1936,9 +2025,7 @@ fn prepare_memory_runtime_layout( "cached runtime catalog profile mismatch" ); let mutable_root = StorageRoot::memory(); - for path in ["/home", "/dev", "/dev/shm", "/tmp"] { - mutable_root.create_dir_all(Path::new(path))?; - } + mutable_root.create_runtime_dirs()?; Ok(RuntimeLayout { catalog_profile: profile, kind: RuntimeLayoutKind::SharedRuntimeOverlay, diff --git a/src/wasix/sdks/rust/src/oliphaunt/data_dir.rs b/src/wasix/sdks/rust/src/oliphaunt/data_dir.rs index fd5d294bd..fa5f0a5dc 100644 --- a/src/wasix/sdks/rust/src/oliphaunt/data_dir.rs +++ b/src/wasix/sdks/rust/src/oliphaunt/data_dir.rs @@ -16,7 +16,7 @@ use super::base::DirectoryLock; use super::database_root_descriptor::{ PGDATA_DIRECTORY, PHYSICAL_FORMAT, POSTGRES_MAJOR, write_database_root_descriptor, }; -use crate::oliphaunt::storage::{PgDataStorage, vfs_read}; +use crate::oliphaunt::storage::{PgDataStorage, for_each_publication_file, vfs_read}; use crate::{StorageCommitState, StorageErrorCode, StorageErrorPhase}; const PHYSICAL_ARCHIVE_MANIFEST_NAME: &str = ".oliphaunt/backup-manifest.properties"; @@ -49,6 +49,7 @@ enum HostTreeEntry { enum VirtualEntrySource { Directory, File(Vec), + InitializedFile(PathBuf), } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -211,13 +212,45 @@ pub(crate) fn materialize_virtual_pgdata_view( filesystem: &(dyn VirtualFileSystem + Send + Sync), ) -> Result { let temp = tempfile::TempDir::new().context("create materialized virtual PGDATA view")?; + copy_virtual_pgdata_into(filesystem, temp.path(), false)?; + Ok(temp) +} + +pub(crate) fn materialize_initialized_pgdata_into( + filesystem: &(dyn VirtualFileSystem + Send + Sync), + destination_root: &Path, +) -> Result<()> { + // A completed initdb needs its entire WAL tree; an online backup's bulk + // view intentionally excludes WAL until its final checkpoint is captured. + copy_virtual_pgdata_into(filesystem, destination_root, true) +} + +fn copy_virtual_pgdata_into( + filesystem: &(dyn VirtualFileSystem + Send + Sync), + destination_root: &Path, + initialized: bool, +) -> Result<()> { let mut entries = BTreeMap::::new(); - collect_virtual_pgdata_entries(filesystem, Path::new("/"), Path::new("/"), &mut entries)?; + collect_virtual_pgdata_entries( + filesystem, + Path::new("/"), + Path::new("/"), + &mut entries, + initialized, + )?; + let mut files = Vec::new(); for (relative, source) in entries { - let destination = temp.path().join(relative); + let destination = destination_root.join(relative); match source { - VirtualEntrySource::Directory => fs::create_dir_all(&destination) - .with_context(|| format!("create {}", destination.display()))?, + VirtualEntrySource::Directory => { + if initialized { + create_private_directory(&destination)?; + } else { + fs::create_dir_all(&destination) + .with_context(|| format!("create {}", destination.display()))?; + } + } + VirtualEntrySource::InitializedFile(path) => files.push((destination, path)), VirtualEntrySource::File(bytes) => { if let Some(parent) = destination.parent() { fs::create_dir_all(parent)?; @@ -227,7 +260,21 @@ pub(crate) fn materialize_virtual_pgdata_view( } } } - Ok(temp) + // Initdb has exited, so its private filesystem is no longer changing. + // Read and copy each file in a worker instead of buffering the entire tree. + // Finish every file before the caller checks syncs and renames staging. Backups + // retain their sequential bulk-copy path and separate final WAL capture. + for_each_publication_file(&files, |(destination, source)| { + let bytes = vfs_read(filesystem, source)?; + let mut file = create_private_file(destination) + .with_context(|| format!("create {}", destination.display()))?; + file.write_all(&bytes) + .with_context(|| format!("write {}", destination.display()))?; + apply_private_permissions(destination, 0o600)?; + super::base::pre_sync_publication_file(&file); + Ok(()) + })?; + Ok(()) } pub(crate) fn refresh_materialized_pg_control( @@ -1104,6 +1151,7 @@ fn collect_virtual_pgdata_entries( root: &Path, current: &Path, entries: &mut BTreeMap, + initialized: bool, ) -> Result<()> { let mut children = filesystem .read_dir(current) @@ -1118,7 +1166,9 @@ fn collect_virtual_pgdata_entries( .strip_prefix(root) .with_context(|| format!("strip virtual PGDATA root {}", root.display()))? .to_path_buf(); - if relative.as_os_str().is_empty() || should_skip_bulk_backup_entry(&relative) { + if relative.as_os_str().is_empty() + || (!initialized && should_skip_bulk_backup_entry(&relative)) + { continue; } let metadata = child @@ -1126,11 +1176,15 @@ fn collect_virtual_pgdata_entries( .with_context(|| format!("stat virtual PGDATA entry {}", path.display()))?; if metadata.is_dir() { entries.insert(relative.clone(), VirtualEntrySource::Directory); - collect_virtual_pgdata_entries(filesystem, root, &path, entries)?; + collect_virtual_pgdata_entries(filesystem, root, &path, entries, initialized)?; } else if metadata.is_file() { entries.insert( relative, - VirtualEntrySource::File(vfs_read(filesystem, &path)?), + if initialized { + VirtualEntrySource::InitializedFile(path) + } else { + VirtualEntrySource::File(vfs_read(filesystem, &path)?) + }, ); } else { bail!( @@ -1905,6 +1959,53 @@ mod tests { Ok(()) } + #[test] + fn initialized_virtual_pgdata_keeps_wal_and_empty_directories() -> Result<()> { + let filesystem = virtual_fs::mem_fs::FileSystem::default(); + super::super::storage::vfs_create_dir_all( + &filesystem, + Path::new("/pg_wal/archive_status"), + )?; + super::super::storage::vfs_create_dir_all(&filesystem, Path::new("/base/1"))?; + super::super::storage::vfs_write(&filesystem, Path::new("/pg_wal/segment"), b"wal")?; + super::super::storage::vfs_write(&filesystem, Path::new("/base/1/catalog"), b"catalog")?; + let destination = tempfile::tempdir()?; + + materialize_initialized_pgdata_into(&filesystem, destination.path())?; + + assert_eq!(fs::read(destination.path().join("pg_wal/segment"))?, b"wal"); + assert_eq!( + fs::read(destination.path().join("base/1/catalog"))?, + b"catalog" + ); + assert!(destination.path().join("pg_wal/archive_status").is_dir()); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + fs::metadata(destination.path().join("base/1"))? + .permissions() + .mode() + & 0o777, + 0o700 + ); + assert_eq!( + fs::metadata(destination.path().join("base/1/catalog"))? + .permissions() + .mode() + & 0o777, + 0o600 + ); + } + let backup = materialize_virtual_pgdata_view(&filesystem)?; + assert!(!backup.path().join("pg_wal/segment").exists()); + assert_eq!(fs::read(backup.path().join("base/1/catalog"))?, b"catalog"); + let blocked = tempfile::tempdir()?; + fs::create_dir_all(blocked.path().join("base/1/catalog"))?; + assert!(materialize_initialized_pgdata_into(&filesystem, blocked.path()).is_err()); + Ok(()) + } + #[test] fn final_pg_control_refresh_replaces_the_bulk_snapshot() -> Result<()> { let source = tempfile::tempdir()?; diff --git a/src/wasix/sdks/rust/src/oliphaunt/postgres_mod.rs b/src/wasix/sdks/rust/src/oliphaunt/postgres_mod.rs index 7d0b7d312..bc7123e27 100644 --- a/src/wasix/sdks/rust/src/oliphaunt/postgres_mod.rs +++ b/src/wasix/sdks/rust/src/oliphaunt/postgres_mod.rs @@ -46,6 +46,8 @@ const PGDATA_DIR: &str = "/base"; const ICU_DATA_DIR: &str = "/share/icu"; const SKIP_ICU_COLLATION_DISCOVERY_ENV: &str = "OLIPHAUNT_INTERNAL_SKIP_ICU_DISCOVERY"; const WASM_PREFIX: &str = "/"; +const DEFAULT_SHARED_BUFFERS: &str = "128MB"; +const PRIVATE_INITDB_SHARED_BUFFERS: &str = "8MB"; const RUNTIME_SIDE_MODULES: &[(&str, &str)] = &[ ("plpgsql.so", "runtime-support:plpgsql"), ("dict_snowball.so", "runtime-support:dict_snowball"), @@ -1006,6 +1008,9 @@ fn build_wasix_runtime( } fn run_split_initdb(runtime_layout: &RuntimeLayout, pgdata_storage: &PgDataStorage) -> Result<()> { + let filesystem = pgdata_storage + .memory_filesystem() + .context("split initdb requires private memory storage")?; let initdb_module = runtime_layout.module_root.join("bin/initdb"); let postgres_module = runtime_layout.module_root.join("bin/postgres"); ensure!( @@ -1019,11 +1024,6 @@ fn run_split_initdb(runtime_layout: &RuntimeLayout, pgdata_storage: &PgDataStora postgres_module.display() ); - if let PgDataStorage::HostDirectory(pgdata) = pgdata_storage { - fs::create_dir_all(pgdata) - .with_context(|| format!("create fresh PGDATA {}", pgdata.display()))?; - } - let (engine, _) = aot::load_runtime_module()?; let process_runtime = process_wasix_runtime(&engine)?; seed_wasix_module_cache( @@ -1116,9 +1116,41 @@ fn run_split_initdb(runtime_layout: &RuntimeLayout, pgdata_storage: &PgDataStora }); } } + let path = Path::new("/postgresql.conf"); + let text = String::from_utf8(super::storage::vfs_read(filesystem.as_ref(), path)?) + .context("decode private initdb configuration")?; + let restored = restore_private_initdb_configuration(&text)?; + super::storage::vfs_write(filesystem.as_ref(), path, restored.as_bytes())?; Ok(()) } +fn restore_private_initdb_configuration(text: &str) -> Result { + let temporary = format!("shared_buffers = {PRIVATE_INITDB_SHARED_BUFFERS}"); + let replacement = format!("shared_buffers = {DEFAULT_SHARED_BUFFERS}"); + let mut restored = String::with_capacity(text.len() + 2); + let mut found = false; + for line in text.split_inclusive('\n') { + if let Some(tail) = line.strip_prefix(&temporary) + && (tail.trim().is_empty() || tail.trim_start().starts_with('#')) + { + ensure!( + !found, + "private initdb emitted duplicate shared_buffers settings" + ); + restored.push_str(&replacement); + restored.push_str(tail); + found = true; + } else { + restored.push_str(line); + } + } + ensure!( + found, + "private initdb did not emit its temporary shared_buffers setting" + ); + Ok(restored) +} + fn split_initdb_root_filesystem( runtime_layout: &RuntimeLayout, pgdata_storage: &PgDataStorage, @@ -1263,8 +1295,8 @@ fn dir_entry_sample(path: &Path) -> String { } } -fn split_initdb_args() -> Vec<&'static str> { - vec![ +fn split_initdb_args() -> Vec { + let mut args = vec![ "--allow-group-access", "--encoding", "UTF8", @@ -1277,6 +1309,17 @@ fn split_initdb_args() -> Vec<&'static str> { "-D", PGDATA_DIR, ] + .into_iter() + .map(str::to_owned) + .collect::>(); + // Bootstrap is private memory. Restore the serving buffer size before + // publication, whose checked syncs cover the complete initialized tree. + args.extend([ + "--no-sync".to_owned(), + "--set".to_owned(), + format!("shared_buffers={PRIVATE_INITDB_SHARED_BUFFERS}"), + ]); + args } fn split_initdb_binary_package( @@ -1799,16 +1842,14 @@ const DEFAULT_STARTUP_GUCS: &[(&str, &str)] = &[ ("log_checkpoints", "false"), ("wal_buffers", "4MB"), ("min_wal_size", "80MB"), - ("shared_buffers", "128MB"), + ("shared_buffers", DEFAULT_SHARED_BUFFERS), ]; fn ensure_runtime_dirs( runtime_storage: &StorageRoot, pgdata_storage: &PgDataStorage, ) -> Result<()> { - for path in ["/", "/home", "/dev", "/dev/shm", "/tmp"] { - runtime_storage.create_dir_all(Path::new(path))?; - } + runtime_storage.create_runtime_dirs()?; if let PgDataStorage::HostDirectory(pgdata) = pgdata_storage { fs::create_dir_all(pgdata) .with_context(|| format!("create PGDATA {}", pgdata.display()))?; @@ -2027,6 +2068,43 @@ mod tests { ); } + #[test] + fn private_initdb_settings_do_not_escape_bootstrap() -> Result<()> { + let args = split_initdb_args(); + assert!(args.iter().any(|arg| arg == "--no-sync")); + assert!(args.iter().any(|arg| arg == "shared_buffers=8MB")); + let text = "# shared_buffers = 8MB\nshared_buffers = 8MB\t# min 128kB\nfsync = on\n"; + assert_eq!( + restore_private_initdb_configuration(text)?, + "# shared_buffers = 8MB\nshared_buffers = 128MB\t# min 128kB\nfsync = on\n" + ); + assert!(restore_private_initdb_configuration("# shared_buffers = 8MB\n").is_err()); + assert!(restore_private_initdb_configuration("shared_buffers = 8MB0\n").is_err()); + assert!( + restore_private_initdb_configuration("shared_buffers = 8MB\nshared_buffers = 8MB\n") + .is_err() + ); + assert_eq!( + restore_private_initdb_configuration("shared_buffers = 8MB")?, + "shared_buffers = 128MB" + ); + let layout = RuntimeLayout { + catalog_profile: crate::CatalogProfile::Standard, + kind: super::super::base::RuntimeLayoutKind::FullLocal, + mutable_root: StorageRoot::memory(), + shared_root: None, + module_root: "missing-runtime".into(), + }; + let host = PgDataStorage::host_directory("must-not-initialize"); + let error = run_split_initdb(&layout, &host).expect_err("host initdb must be rejected"); + assert!( + error + .to_string() + .contains("requires private memory storage") + ); + Ok(()) + } + #[test] fn startup_error_summary_includes_postgres_fields() { let response = oliphaunt_query::wire::error_response( diff --git a/src/wasix/sdks/rust/src/oliphaunt/storage.rs b/src/wasix/sdks/rust/src/oliphaunt/storage.rs index a2d04baa7..642c4ac52 100644 --- a/src/wasix/sdks/rust/src/oliphaunt/storage.rs +++ b/src/wasix/sdks/rust/src/oliphaunt/storage.rs @@ -107,6 +107,13 @@ impl StorageRoot { } } + pub(super) fn create_runtime_dirs(&self) -> Result<()> { + for path in ["/", "/home", "/dev", "/dev/shm", "/tmp"] { + self.create_dir_all(Path::new(path))?; + } + Ok(()) + } + #[cfg(feature = "extensions")] pub(crate) fn read(&self, path: &Path) -> Result> { match self { @@ -226,11 +233,38 @@ pub(crate) fn vfs_read( .open(path) .with_context(|| format!("open virtual file {} for reading", path.display()))?; let mut bytes = Vec::new(); + bytes + .try_reserve_exact(usize::try_from(file.size()).context("virtual file is too large")?) + .context("reserve virtual file read buffer")?; block_on_vfs(file.read_to_end(&mut bytes)) .with_context(|| format!("read virtual file {}", path.display()))?; Ok(bytes) } +pub(super) fn for_each_publication_file( + files: &[T], + operation: impl Fn(&T) -> Result<()> + Sync, +) -> Result<()> { + const WORKERS: usize = 4; + std::thread::scope(|scope| { + let mut workers = Vec::new(); + for chunk in files.chunks(files.len().div_ceil(WORKERS).max(1)) { + let operation = &operation; + workers.push( + std::thread::Builder::new() + .spawn_scoped(scope, move || chunk.iter().try_for_each(operation)) + .context("start publication worker")?, + ); + } + for worker in workers { + worker + .join() + .unwrap_or_else(|payload| std::panic::resume_unwind(payload))?; + } + Ok(()) + }) +} + pub(crate) fn vfs_file_exists(filesystem: &(dyn FileSystem + Send + Sync), path: &Path) -> bool { filesystem .metadata(path) @@ -252,6 +286,23 @@ mod tests { use super::*; use tempfile::TempDir; + #[test] + fn publication_workers_finish_before_returning_an_error() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let finished = AtomicUsize::new(0); + let result = for_each_publication_file(&[0, 1, 2, 3], |index| { + if *index == 0 { + bail!("publication failed"); + } + finished.fetch_add(1, Ordering::SeqCst); + Ok(()) + }); + assert!(result.is_err()); + assert_eq!(finished.load(Ordering::SeqCst), 3); + assert!(for_each_publication_file::<()>(&[], |_| panic!("empty input")).is_ok()); + } + #[test] fn host_and_memory_storage_share_the_file_contract() -> Result<()> { let directory = TempDir::new()?; diff --git a/src/wasix/sdks/rust/tests/runtime_smoke.rs b/src/wasix/sdks/rust/tests/runtime_smoke.rs index 2da39f3dd..d3ee6be00 100644 --- a/src/wasix/sdks/rust/tests/runtime_smoke.rs +++ b/src/wasix/sdks/rust/tests/runtime_smoke.rs @@ -30,10 +30,26 @@ fn direct_api_query_transaction_persistence_and_backup() -> Result<()> { let workspace = tempfile::TempDir::new()?; let source_root = workspace.path().join("source"); let mut database = Oliphaunt::builder() - .storage(DatabaseStorage::Directory(source_root)) + .storage(DatabaseStorage::Directory(source_root.clone())) .startup_guc("work_mem", "8MB") .open()?; + let settings = database.query( + "SELECT current_setting('shared_buffers') AS buffers, current_setting('fsync') AS fsync, \ + current_setting('full_page_writes') AS full_pages, \ + current_setting('synchronous_commit') AS synchronous_commit", + )?; + assert_eq!(settings.get_text(0, "buffers")?, Some("128MB")); + for setting in ["fsync", "full_pages", "synchronous_commit"] { + assert_eq!(settings.get_text(0, setting)?, Some("on")); + } + let configuration = std::fs::read_to_string(source_root.join("pgdata/postgresql.conf"))?; + assert!( + configuration + .lines() + .any(|line| line.starts_with("shared_buffers = 128MB")) + ); + database.execute("CREATE TABLE items(id integer PRIMARY KEY, value text NOT NULL)")?; let multiple = database .execute("CREATE TABLE must_not_exist(id integer); INSERT INTO must_not_exist VALUES (1)") @@ -140,7 +156,15 @@ fn direct_api_query_transaction_persistence_and_backup() -> Result<()> { #[test] fn direct_api_recovers_after_postgres_error() -> Result<()> { - let mut database = Oliphaunt::open()?; + let mut database = Oliphaunt::builder() + .startup_guc("shared_buffers", "16MB") + .open()?; + assert_eq!( + database + .query("SHOW shared_buffers")? + .get_text(0, "shared_buffers")?, + Some("16MB") + ); let error = database .query_with_params("SELECT 1 / $1::int4", [0_i32]) .expect_err("division by zero must fail"); diff --git a/tools/ci/workflow-moon-transfers.test.mts b/tools/ci/workflow-moon-transfers.test.mts index 450313ec7..2338df792 100644 --- a/tools/ci/workflow-moon-transfers.test.mts +++ b/tools/ci/workflow-moon-transfers.test.mts @@ -129,11 +129,29 @@ if (!process.env.OLIPHAUNT_TRANSFER_FIXTURE_PHASE) assert(dependencies.has('checks'), `${id} can start before source checks`); assert(dependencies.has('tests'), `${id} can start before source tests`); const direct = [job.needs ?? []].flat(); - if (direct.includes('checks') && direct.includes('tests') && direct.length === 3) + if (direct.includes('checks') && direct.includes('tests') && direct.length === 3) { + // Optional source matrices may be skipped; require the aggregate gate results. assert( - !/always\(|!cancelled\(/u.test(job.if ?? ''), + job.if?.startsWith( + "${{ !cancelled() && needs.affected.result == 'success' && needs.checks.result == 'success' && needs.tests.result == 'success' && ", + ), `${id} must require successful source gates`, ); + } + } + }); + +if (!process.env.OLIPHAUNT_TRANSFER_FIXTURE_PHASE) + test('WASIX aggregates require successful selected hosts before accepting their artifacts', () => { + for (const id of ['wasix-napi', 'liboliphaunt-wasix-aot']) { + for (const host of ['linux', 'other']) { + assert( + workflow.jobs[id].if.includes( + `(fromJson(needs.affected.outputs.liboliphaunt_wasix_aot_runtime_matrix_${host}).include[0] == null || needs.wasix-host-${host}.result == 'success')`, + ), + `${id} accepts a skipped selected ${host} host`, + ); + } } });