Merge remote-tracking branch 'origin/master' into bfops/smoketest-flakes

Author: bfops

.cargo/config.toml

@@ -5,6 +5,7 @@ rustflags = ["--cfg", "tokio_unstable"]
 bump-versions = "run -p upgrade-version --"
 llm = "run --package xtask-llm-benchmark --bin llm_benchmark --"
 ci = "run -p ci --"
+regen = "run -p regen --"
 smoketest = "ci smoketests --"
 smoketests = "smoketest"
 lint = "ci lint --"

.github/workflows/ci.yml

@@ -689,7 +689,7 @@ jobs:
           }
 
       - name: Hydrate Unity SDK DLLs
-        run: cargo ci dlls
+        run: cargo regen csharp dlls
 
       - name: Check Unity meta files
         uses: DeNA/unity-meta-check@v3
@@ -830,10 +830,9 @@ jobs:
           ln -sf $CARGO_HOME/bin/spacetimedb-cli $CARGO_HOME/bin/spacetime
 
       - name: Check quickstart-chat bindings are up to date
-        working-directory: sdks/csharp
         run: |
-          bash tools~/gen-quickstart.sh
-          "${GITHUB_WORKSPACE}"/tools/check-diff.sh examples~/quickstart-chat || {
+          bash sdks/csharp/tools~/gen-quickstart.sh
+          tools/check-diff.sh templates/chat-console-cs/module_bindings || {
               echo 'Error: quickstart-chat bindings have changed. Please run `sdks/csharp/tools~/gen-quickstart.sh`.'
               exit 1
           }

Cargo.lock

@@ -65,6 +65,7 @@ members = [
   "tools/replace-spacetimedb",
   "tools/generate-client-api",
   "tools/gen-bindings",
+  "tools/regen",
   "tools/xtask-llm-benchmark",
   "crates/bindings-typescript/test-app/server",
   "crates/bindings-typescript/test-react-router-app/server",

Cargo.toml

@@ -240,6 +240,8 @@ enum Error {
     },
     #[error("encountered failed transaction: {reason}")]
     TransactionFailure { reason: Box<str> },
+    #[error("encountered error in initial subscribe: {reason}")]
+    SubscribeFailure { reason: Box<str> },
     #[error("error formatting response: {source:#}")]
     Reformat {
         #[source]
@@ -295,6 +297,9 @@ where
                 }
                 break;
             }
+            ws_v1::ServerMessage::SubscriptionError(error) => {
+                return Err(Error::SubscribeFailure { reason: error.error });
+            }
             ws_v1::ServerMessage::TransactionUpdate(ws_v1::TransactionUpdate { status, .. }) => {
                 return Err(match status {
                     ws_v1::UpdateStatus::Failed(msg) => Error::TransactionFailure { reason: msg },
@@ -341,6 +346,9 @@ where
                     details: "received a second initial subscription update",
                 })
             }
+            ws_v1::ServerMessage::SubscriptionError(error) => {
+                return Err(Error::SubscribeFailure { reason: error.error });
+            }
             ws_v1::ServerMessage::TransactionUpdateLight(ws_v1::TransactionUpdateLight { update, .. })
             | ws_v1::ServerMessage::TransactionUpdate(ws_v1::TransactionUpdate {
                 status: ws_v1::UpdateStatus::Committed(update),

crates/cli/src/subcommands/subscribe.rs

@@ -9,6 +9,8 @@ use log::trace;
 use repo::{fs::OnNewSegmentFn, Repo};
 use spacetimedb_paths::server::CommitLogDir;
 
+pub use spacetimedb_fs_utils::compression::CompressionStats;
+
 pub mod commit;
 pub mod commitlog;
 mod index;
@@ -330,16 +332,18 @@ impl<T> Commitlog<T> {
     }
 
     /// Compress the segments at the offsets provided, marking them as immutable.
-    pub fn compress_segments(&self, offsets: &[u64]) -> io::Result<()> {
+    pub fn compress_segments(&self, offsets: &[u64]) -> io::Result<CompressionStats> {
         // even though `compress_segment` takes &self, we take an
         // exclusive lock to avoid any weirdness happening.
         #[allow(clippy::readonly_write_lock)]
         let inner = self.inner.write().unwrap();
         assert!(!offsets.contains(&inner.head.min_tx_offset()));
         // TODO: parallelize, maybe
-        offsets
-            .iter()
-            .try_for_each(|&offset| inner.repo.compress_segment(offset))
+        let mut stats = <_>::default();
+        for offset in offsets {
+            stats += inner.repo.compress_segment(*offset)?;
+        }
+        Ok(stats)
     }
 
     /// Remove all data from the log and reopen it.

crates/commitlog/src/lib.rs

@@ -4,14 +4,14 @@ use std::io;
 use std::sync::Arc;
 
 use log::{debug, warn};
-use spacetimedb_fs_utils::compression::{compress_with_zstd, CompressReader};
+use spacetimedb_fs_utils::compression::{CompressReader, CompressionStats};
 use spacetimedb_fs_utils::lockfile;
 use spacetimedb_paths::server::{CommitLogDir, SegmentFile};
 use tempfile::NamedTempFile;
 
-use crate::segment::{self, FileLike};
-
 use super::{Repo, SegmentLen, SegmentReader, TxOffset, TxOffsetIndex, TxOffsetIndexMut};
+use crate::repo::CompressOnce;
+use crate::segment::{self, FileLike};
 
 const SEGMENT_FILE_EXT: &str = ".stdb.log";
 
@@ -317,21 +317,18 @@ impl Repo for Fs {
         fs::remove_file(self.segment_path(offset))
     }
 
-    fn compress_segment(&self, offset: u64) -> io::Result<()> {
+    fn compress_segment_with(&self, offset: u64, f: impl CompressOnce) -> io::Result<CompressionStats> {
         let src = self.open_segment_reader(offset)?;
         // if it's already compressed, leave it be
         let CompressReader::None(mut src) = src.inner else {
-            return Ok(());
+            return Ok(<_>::default());
         };
 
         let mut dst = NamedTempFile::new_in(&self.root)?;
-        // bytes per frame. in the future, it might be worth looking into putting
-        // every commit into its own frame, to make seeking more efficient.
-        let max_frame_size = 0x1000;
-        compress_with_zstd(&mut src, &mut dst, Some(max_frame_size))?;
+        let stats = f.compress(&mut src, &mut dst)?;
         dst.persist(self.segment_path(offset))?;
 
-        Ok(())
+        Ok(stats)
     }
 
     fn existing_offsets(&self) -> io::Result<Vec<u64>> {

crates/commitlog/src/repo/fs.rs

@@ -11,6 +11,7 @@ use crate::repo::{
 
 mod segment;
 pub use segment::{ReadOnlySegment, Segment};
+use spacetimedb_fs_utils::compression::CompressionStats;
 
 pub const PAGE_SIZE: usize = 4096;
 
@@ -105,8 +106,8 @@ impl Repo for Memory {
         Ok(())
     }
 
-    fn compress_segment(&self, _offset: u64) -> io::Result<()> {
-        Ok(())
+    fn compress_segment_with(&self, _: u64, _: impl super::CompressOnce) -> io::Result<CompressionStats> {
+        Ok(<_>::default())
     }
 
     fn existing_offsets(&self) -> io::Result<Vec<u64>> {

crates/commitlog/src/repo/mem.rs

@@ -4,6 +4,8 @@ use std::{
 };
 
 use log::{debug, warn};
+use spacetimedb_fs_utils::compression::Zstd;
+pub use spacetimedb_fs_utils::compression::{CompressOnce, CompressionStats};
 
 use crate::{
     commit::Commit,
@@ -121,7 +123,13 @@ pub trait Repo: Clone + fmt::Display {
     fn remove_segment(&self, offset: u64) -> io::Result<()>;
 
     /// Compress a segment in storage, marking it as immutable.
-    fn compress_segment(&self, offset: u64) -> io::Result<()>;
+    fn compress_segment(&self, offset: u64) -> io::Result<CompressionStats> {
+        self.compress_segment_with(offset, segment_compressor())
+    }
+
+    /// Compress a segment using a supplied [CompressOnce], marking it as
+    /// immutable.
+    fn compress_segment_with(&self, offset: u64, f: impl CompressOnce) -> io::Result<CompressionStats>;
 
     /// Traverse all segments in this repository and return list of their
     /// offsets, sorted in ascending order.
@@ -164,8 +172,8 @@ impl<T: Repo> Repo for &T {
         T::remove_segment(self, offset)
     }
 
-    fn compress_segment(&self, offset: u64) -> io::Result<()> {
-        T::compress_segment(self, offset)
+    fn compress_segment_with(&self, offset: u64, f: impl CompressOnce) -> io::Result<CompressionStats> {
+        T::compress_segment_with(self, offset, f)
     }
 
     fn existing_offsets(&self) -> io::Result<Vec<u64>> {
@@ -408,6 +416,17 @@ pub fn open_segment_reader<R: Repo>(
         .map_err(|source| with_segment_context("reading segment header", repo, offset, source))
 }
 
+/// Obtain the canonical [CompressOnce] compressor for segments.
+///
+/// The compressor will create seekable [Zstd] archives with a max frame size
+/// of 4KiB. That is, seeking to an arbitrary byte offset (of the uncompressed
+/// segment) within the archive will decompress 4KiB of data on average.
+pub fn segment_compressor() -> Zstd {
+    Zstd {
+        max_frame_size: Some(0x1000),
+    }
+}
+
 fn segment_label<R: Repo>(repo: &R, offset: u64) -> String {
     repo.segment_file_path(offset)
         .unwrap_or_else(|| format!("offset {offset}"))

crates/commitlog/src/repo/mod.rs

@@ -8,6 +8,7 @@ use std::{
 
 use log::{debug, info};
 use pretty_assertions::assert_matches;
+use spacetimedb_fs_utils::compression::CompressionStats;
 
 use crate::{
     commitlog, payload,
@@ -295,8 +296,8 @@ impl Repo for ShortMem {
         self.inner.remove_segment(offset)
     }
 
-    fn compress_segment(&self, offset: u64) -> io::Result<()> {
-        self.inner.compress_segment(offset)
+    fn compress_segment_with(&self, offset: u64, f: impl repo::CompressOnce) -> io::Result<CompressionStats> {
+        self.inner.compress_segment_with(offset, f)
     }
 
     fn existing_offsets(&self) -> io::Result<Vec<u64>> {