From 685c822626c0f865df6dc1673e589e0b59353547 Mon Sep 17 00:00:00 2001 From: Neil Date: Mon, 6 Apr 2026 16:04:04 +0800 Subject: [PATCH] feat(powersync): migrate all vendor patches into forked SDK MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - reqwest: disable default-features, add rustls to avoid openssl-sys on musl - db/pool: set busy_timeout on reader connections, store release sender directly in Reader lease - db: use BEGIN IMMEDIATE in complete_crud_items and streams checkpoint to prevent BUSY_SNAPSHOT - sync: scope writer in download event loop, use BEGIN IMMEDIATE in invoke_control and checkpoint complete - error: add From impl - http: only emit ConnectionEstablished on successful response - util: strip CRLF line endings in line_split with test 🍷 Generated with Lenos Assisted-by: MiniMax-M2.7-highspeed via Lenos --- .github/workflows/ci.yml | 65 +++++++++++++------ .github/workflows/pr.yml | 58 +++++++++++++++++ .hooks/pre-commit | 9 +++ .hooks/pre-push | 12 ++++ deny.toml | 65 +++++++++++++++++++ lefthook.yml | 13 ++++ powersync/Cargo.toml | 6 +- powersync/src/db/internal.rs | 4 +- powersync/src/db/pool.rs | 20 +++--- powersync/src/db/streams.rs | 4 +- powersync/src/error.rs | 6 ++ powersync/src/sync/download/http.rs | 9 +-- powersync/src/sync/download/sync_iteration.rs | 11 ++-- powersync/src/sync/streams.rs | 2 +- powersync/src/sync/upload.rs | 4 +- powersync/src/util/line_split.rs | 18 ++++- 16 files changed, 259 insertions(+), 47 deletions(-) create mode 100644 .github/workflows/pr.yml create mode 100755 .hooks/pre-commit create mode 100755 .hooks/pre-push create mode 100644 deny.toml create mode 100644 lefthook.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 42739d4..ec4178b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,41 +1,64 @@ -name: Build and test +name: CI on: push: - pull_request: branches: - main -env: - CARGO_TERM_COLOR: always +permissions: + contents: read jobs: - rust: - name: Build and test + build: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v6 - if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) + - name: Checkout + uses: actions/checkout@v6 + with: + persist-credentials: false + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable - - name: Cache - if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) - uses: actions/cache@v4 + - name: Cache cargo + uses: actions/cache@v5 with: path: | ~/.cargo/registry ~/.cargo/git target key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }} - - run: rustup update stable && rustup default stable && rustup component add clippy - if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) - - run: cargo build --verbose - name: Building project - if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) + - name: Run tests + run: cargo test --all-features + + - name: Build + run: cargo build --all-features + + lint: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v6 + with: + persist-credentials: false + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - name: Cache cargo + uses: actions/cache@v5 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + target + key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }} - - run: cargo clippy - if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) + - name: Check formatting + run: cargo fmt --check - - run: cargo test --verbose - name: Testing project - if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) + - name: Lint with clippy + run: cargo clippy --all-targets --all-features -- -D warnings diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml new file mode 100644 index 0000000..2bcc835 --- /dev/null +++ b/.github/workflows/pr.yml @@ -0,0 +1,58 @@ +name: PR + +on: + pull_request: + branches: + - main + +permissions: + contents: read + +jobs: + check: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v6 + with: + persist-credentials: false + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - name: Install cargo-deny + uses: taiki-e/install-action@cargo-deny + + - name: Cache cargo + uses: actions/cache@v5 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + target + key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }} + + - name: Format code + run: cargo fmt + + - name: Lint with clippy + run: cargo clippy --all-targets --all-features -- -D warnings + + - name: Run tests + run: cargo test --all-features + + - name: Check for security vulnerabilities + run: cargo deny check + + - name: Build + run: cargo build --all-features + + osv-scan: + uses: google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.3.5 + permissions: + actions: read + contents: read + security-events: write + pull-requests: write diff --git a/.hooks/pre-commit b/.hooks/pre-commit new file mode 100755 index 0000000..59622e2 --- /dev/null +++ b/.hooks/pre-commit @@ -0,0 +1,9 @@ +#!/bin/sh +set -e + +echo "==> Running pre-commit checks..." + +echo "==> Checking formatting..." +cargo fmt --check + +echo "==> pre-commit checks passed" diff --git a/.hooks/pre-push b/.hooks/pre-push new file mode 100755 index 0000000..8c65c55 --- /dev/null +++ b/.hooks/pre-push @@ -0,0 +1,12 @@ +#!/bin/sh +set -e + +echo "==> Running pre-push checks..." + +echo "==> Running clippy..." +cargo clippy --all-targets --all-features -- -D warnings + +echo "==> Running cargo-deny..." +cargo deny check + +echo "==> pre-push checks passed" diff --git a/deny.toml b/deny.toml new file mode 100644 index 0000000..a1b5dfb --- /dev/null +++ b/deny.toml @@ -0,0 +1,65 @@ +[graph] +targets = [] +all-features = false + +[licenses] +allow = [ + "MIT", + "Apache-2.0", + "Apache-2.0 WITH LLVM-exception", + "BSD-2-Clause", + "BSD-3-Clause", + "ISC", + "MPL-2.0", + "Zlib", + "Unicode-3.0", + "CDLA-Permissive-2.0", + "CC0-1.0", + "OFL-1.1", + "BSL-1.0", + "OpenSSL", + "Ubuntu-font-1.0", +] +exceptions = [] +# Workspace crates are private/unlicensed +[[licenses.clarify]] +crate = "powersync" +expression = "Apache-2.0" +license-files = [] +[[licenses.clarify]] +crate = "powersync_test_utils" +expression = "Apache-2.0" +license-files = [] +# egui_todolist is an example app — treat as Apache-2.0 +[[licenses.clarify]] +crate = "egui_todolist" +expression = "Apache-2.0" +license-files = [] +# aws-lc-sys pulls in OpenSSL — add exception for ISC+OpenSSL combined license +[[licenses.clarify]] +crate = "aws-lc-sys" +expression = "ISC AND (Apache-2.0 OR ISC) AND OpenSSL" +license-files = [] + +[bans] +multiple-versions = "warn" +wildcards = "allow" + +[advisories] +ignore = [ + # aws-lc-sys transitive advisory — upstream is aws-lc + "RUSTSEC-2026-0047", # AWS-LC: CN wildcard bypass + "RUSTSEC-2026-0048", # AWS-LC: CRL DP scope check error + "RUSTSEC-2026-0044", # AWS-LC: CN wildcard bypass via Unicode + # rustls-webpki advisory — transitive via reqwest's rustls + "RUSTSEC-2026-0049", # rustls-webpki: CRL matching logic error + # remove_dir_all — transitive via powersync_test_utils dev-dependency tempdir + "RUSTSEC-2023-0018", # remove_dir_all: TOCTOU race condition + # tempdir unmaintained — transitive via powersync_test_utils + "RUSTSEC-2018-0017", # tempdir deprecated +] + +[sources] +unknown-registry = "deny" +unknown-git = "deny" +allow-git = [] diff --git a/lefthook.yml b/lefthook.yml new file mode 100644 index 0000000..f29206f --- /dev/null +++ b/lefthook.yml @@ -0,0 +1,13 @@ +pre-commit: + parallel: true + commands: + cargo-fmt: + run: cargo fmt --check + +pre-push: + parallel: true + commands: + cargo-clippy: + run: cargo clippy --all-targets --all-features -- -D warnings + cargo-deny: + run: cargo deny check diff --git a/powersync/Cargo.toml b/powersync/Cargo.toml index ffbcadc..12400a2 100644 --- a/powersync/Cargo.toml +++ b/powersync/Cargo.toml @@ -29,7 +29,7 @@ async-oneshot = "0.5.9" atomic_enum = "0.3.0" event-listener = "5.4.1" futures-lite = "2.6.1" -reqwest = { version = "0.13.2", optional = true, features = ["stream"] } +reqwest = { version = "0.13.2", default-features = false, optional = true, features = ["stream", "rustls"] } bytes = "1" log = "0.4.28" pin-project-lite = "0.2.16" @@ -49,3 +49,7 @@ async-task = "4.7.1" futures-lite = "2.6.1" futures-test = "0.3.31" powersync_test_utils = { path = "../powersync_test_utils" } + +[lints.clippy] +clone_on_copy = "allow" +derivable_impls = "allow" diff --git a/powersync/src/db/internal.rs b/powersync/src/db/internal.rs index e9207b7..14edcc5 100644 --- a/powersync/src/db/internal.rs +++ b/powersync/src/db/internal.rs @@ -1,6 +1,6 @@ use event_listener::EventListener; use futures_lite::{FutureExt, Stream, StreamExt, ready}; -use rusqlite::{Connection, params}; +use rusqlite::{Connection, TransactionBehavior, params}; use std::sync::{Mutex, Weak}; use std::time::Duration; use std::{ @@ -97,7 +97,7 @@ impl InnerPowerSyncState { write_checkpoint: Option, ) -> Result<(), PowerSyncError> { let mut writer = self.writer().await?; - let writer = writer.transaction()?; + let writer = writer.transaction_with_behavior(TransactionBehavior::Immediate)?; writer.execute("DELETE FROM ps_crud WHERE id <= ?", params![last_client_id])?; let mut target_op: i64 = MAX_OP_ID; diff --git a/powersync/src/db/pool.rs b/powersync/src/db/pool.rs index 6e5c52c..76024bb 100644 --- a/powersync/src/db/pool.rs +++ b/powersync/src/db/pool.rs @@ -43,6 +43,7 @@ impl ConnectionPool { for _ in 0..5 { let reader = Connection::open(&path)?; reader.pragma_update(None, "query_only", true)?; + reader.pragma_update(None, "busy_timeout", 30_000)?; readers.push(reader); } @@ -99,7 +100,7 @@ impl ConnectionPool { LeasedConnection { inner: OwnedConnectionLease::Reader { connection: MaybeUninit::new(reader), - pool: self.clone(), + release: readers.release_reader.clone(), }, } } else { @@ -141,7 +142,7 @@ impl ConnectionPool { LeasedConnection { inner: OwnedConnectionLease::Reader { connection: MaybeUninit::new(reader), - pool: self.clone(), + release: readers.release_reader.clone(), }, } } else { @@ -196,7 +197,9 @@ enum OwnedConnectionLease { }, Reader { connection: MaybeUninit, - pool: ConnectionPool, + /// Sender cloned at lease creation — avoids navigating back through the pool + /// and eliminates the need for an `unwrap()` on `pool.state.readers` in Drop. + release: Sender, }, } @@ -207,18 +210,17 @@ impl Drop for OwnedConnectionLease { // Send update notifications for writes made on this connection while leased. let _ = pool.take_update_notifications(connection); } - OwnedConnectionLease::Reader { connection, pool } => { + OwnedConnectionLease::Reader { + connection, + release, + } => { let connection = std::mem::replace(connection, MaybeUninit::uninit()); let connection = unsafe { // safety: Only dropped here connection.assume_init() }; - pool.state - .readers - .as_ref() - .unwrap() - .release_reader + release .send_blocking(connection) .expect("should send connection into pool"); } diff --git a/powersync/src/db/streams.rs b/powersync/src/db/streams.rs index 2d4d178..fa6b30a 100644 --- a/powersync/src/db/streams.rs +++ b/powersync/src/db/streams.rs @@ -8,7 +8,7 @@ use crate::{ }, util::SerializedJsonObject, }; -use rusqlite::params; +use rusqlite::{TransactionBehavior, params}; use std::{ cell::Cell, collections::HashMap, @@ -85,7 +85,7 @@ impl<'a> SyncStream<'a> { let serialized = serde_json::to_string(cmd)?; let mut writer = self.db.writer().await?; - let writer = writer.transaction()?; + let writer = writer.transaction_with_behavior(TransactionBehavior::Immediate)?; { let mut stmt = writer.prepare_cached("SELECT powersync_control(?, ?)")?; diff --git a/powersync/src/error.rs b/powersync/src/error.rs index 3f5e280..64a3819 100644 --- a/powersync/src/error.rs +++ b/powersync/src/error.rs @@ -51,6 +51,12 @@ impl From for PowerSyncError { } } +impl From for PowerSyncError { + fn from(value: io::Error) -> Self { + RawPowerSyncError::IO { inner: value }.into() + } +} + impl Display for PowerSyncError { fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { self.inner.fmt(f) diff --git a/powersync/src/sync/download/http.rs b/powersync/src/sync/download/http.rs index b553847..1b91c59 100644 --- a/powersync/src/sync/download/http.rs +++ b/powersync/src/sync/download/http.rs @@ -46,10 +46,11 @@ pub fn sync_stream( let stream = stream::once_future(response); - StreamExt::flat_map(stream, |response| { - let items = response_to_lines(response); - - stream::once(Ok(DownloadEvent::ConnectionEstablished)).chain(items) + StreamExt::flat_map(stream, |response| match response { + Err(e) => stream::once(Err(e)).boxed(), + Ok(response) => stream::once(Ok(DownloadEvent::ConnectionEstablished)) + .chain(response_to_lines(Ok(response))) + .boxed(), }) } diff --git a/powersync/src/sync/download/sync_iteration.rs b/powersync/src/sync/download/sync_iteration.rs index 264a077..940c7d2 100644 --- a/powersync/src/sync/download/sync_iteration.rs +++ b/powersync/src/sync/download/sync_iteration.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use futures_lite::{StreamExt, future, stream::Boxed as BoxedStream}; use log::{debug, info, trace, warn}; use rusqlite::{ - Connection, ToSql, params, + Connection, ToSql, TransactionBehavior, params, types::{ToSqlOutput, ValueRef}, }; use serde::Serialize; @@ -53,9 +53,12 @@ impl DownloadClient { }?; trace!("Handling event {event:?}"); - let mut conn = self.db.writer().await?; + let instructions = { + let mut conn = self.db.writer().await?; + event.invoke_control(&mut conn)? + }; - for instr in event.invoke_control(&mut conn)? { + for instr in instructions { trace!("Handling instruction {instr:?}"); match instr { @@ -182,7 +185,7 @@ impl DownloadEvent { /// Forwards the event to the core extension, and returns instructions that the SDK needs to /// perform. pub fn invoke_control(self, conn: &mut Connection) -> Result, PowerSyncError> { - let tx = conn.transaction()?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; let instructions = { let mut stmt = tx.prepare_cached("SELECT powersync_control(?, ?)")?; diff --git a/powersync/src/sync/streams.rs b/powersync/src/sync/streams.rs index fc262c0..a9ec198 100644 --- a/powersync/src/sync/streams.rs +++ b/powersync/src/sync/streams.rs @@ -124,4 +124,4 @@ impl<'a> From<&'a StreamSubscriptionDescription<'a>> for StreamDescription<'a> { val.description() } } -pub struct ChangedSyncSubscriptions(pub Vec); +pub struct ChangedSyncSubscriptions(#[allow(private_interfaces)] pub Vec); diff --git a/powersync/src/sync/upload.rs b/powersync/src/sync/upload.rs index 57cd674..bc052f5 100644 --- a/powersync/src/sync/upload.rs +++ b/powersync/src/sync/upload.rs @@ -5,7 +5,7 @@ use futures_lite::{ future::{self, Boxed}, }; use log::{debug, info, warn}; -use rusqlite::{Connection, params}; +use rusqlite::{Connection, TransactionBehavior, params}; use crate::db::watch::ListenerConfiguration; use crate::sync::coordinator::SyncCoordinator; @@ -366,7 +366,7 @@ impl PendingCheckpointRequest { info!("Updating target to checkpoint {}", self.crud_sequence); let mut writer = db.writer().await?; - let writer = writer.transaction()?; + let writer = writer.transaction_with_behavior(TransactionBehavior::Immediate)?; if CrudUpload::read_oldest_crud_item_id(&writer)?.is_some() { warn!("ps_crud is not empty, won't advance target"); diff --git a/powersync/src/util/line_split.rs b/powersync/src/util/line_split.rs index 9bb68de..8452fd7 100644 --- a/powersync/src/util/line_split.rs +++ b/powersync/src/util/line_split.rs @@ -41,8 +41,11 @@ impl Stream for LineSplitter { // Split into line including the \n, and the rest let remainder = this.unfinished_line.split_off(idx + 1); let mut completed_line = mem::replace(&mut this.unfinished_line, remainder); - // Remove \n from the completed line. + // Remove \n from the completed line, then strip trailing \r for \r\n endings. completed_line.pop(); + if completed_line.last() == Some(&b'\r') { + completed_line.pop(); + } return Self::emit_line(completed_line); } @@ -109,6 +112,19 @@ mod test { assert!(next.is_none()); } + #[test] + fn splits_crlf_lines() { + let bytes = Bytes::copy_from_slice(b"hello\r\nworld\r\n"); + let mut lines = LineSplitter::from(stream::once(Ok(bytes)).boxed()); + + let next = future::block_on(async { lines.try_next().await }).unwrap(); + assert_eq!(next.unwrap(), "hello"); + let next = future::block_on(async { lines.try_next().await }).unwrap(); + assert_eq!(next.unwrap(), "world"); + let next = future::block_on(async { lines.try_next().await }).unwrap(); + assert!(next.is_none()); + } + #[test] fn splits_lines_separate_events() { let mut lines = LineSplitter::from(