Feature/simple syncdb (#2464)

* simplify db sync on rpc endpoints * switch to patch-db master * update fe for websocket only stream * fix api --------- Co-authored-by: Matt Hill <mattnine@protonmail.com>
2026-03-30 04:01:58 +00:00 · 2023-10-17 09:49:58 -06:00
parent afbab293a8
commit 202695096a
29 changed files with 101 additions and 168 deletions
--- a/backend/src/action.rs
+++ b/backend/src/action.rs
@@ -134,7 +134,7 @@ pub async fn action(
    let manifest = ctx
        .db
        .peek()
-        .await?
+        .await
        .as_package_data()
        .as_idx(&pkg_id)
        .or_not_found(&pkg_id)?
--- a/backend/src/backup/backup_bulk.rs
+++ b/backend/src/backup/backup_bulk.rs
@@ -56,7 +56,7 @@ pub async fn backup_all(
    package_ids: Option<OrdSet<PackageId>>,
    #[arg] password: crate::auth::PasswordType,
 ) -> Result<(), Error> {
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;
    let old_password_decrypted = old_password
        .as_ref()
        .unwrap_or(&password)
@@ -265,7 +265,7 @@ async fn perform_backup(
        }
    }

-    let ui = ctx.db.peek().await?.into_ui().de()?;
+    let ui = ctx.db.peek().await.into_ui().de()?;

    let mut os_backup_file = AtomicFile::new(
        backup_guard.lock().await.as_ref().join("os-backup.cbor"),
--- a/backend/src/backup/mod.rs
+++ b/backend/src/backup/mod.rs
@@ -134,7 +134,7 @@ impl BackupActions {
        let marketplace_url = ctx
            .db
            .peek()
-            .await?
+            .await
            .as_package_data()
            .as_idx(&pkg_id)
            .or_not_found(pkg_id)?
--- a/backend/src/backup/restore.rs
+++ b/backend/src/backup/restore.rs
@@ -310,7 +310,7 @@ async fn assure_restoring(
    let mut insert_packages = BTreeMap::new();

    for id in ids {
-        let peek = ctx.db.peek().await?;
+        let peek = ctx.db.peek().await;

        let model = peek.as_package_data().as_idx(&id);

--- a/backend/src/config/mod.rs
+++ b/backend/src/config/mod.rs
@@ -167,7 +167,7 @@ pub async fn get(
    #[arg(long = "format")]
    format: Option<IoFormat>,
 ) -> Result<ConfigRes, Error> {
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;
    let manifest = db
        .as_package_data()
        .as_idx(&id)
@@ -256,7 +256,7 @@ pub async fn configure(
    id: &PackageId,
    configure_context: ConfigureContext,
 ) -> Result<BTreeMap<PackageId, String>, Error> {
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;
    let package = db
        .as_package_data()
        .as_idx(id)
--- a/backend/src/config/spec.rs
+++ b/backend/src/config/spec.rs
@@ -1696,7 +1696,6 @@ impl TorAddressPointer {
            .db
            .peek()
            .await
-            .map_err(|e| ConfigurationError::SystemError(e))?
            .as_package_data()
            .as_idx(&self.package_id)
            .and_then(|pde| pde.as_installed())
@@ -1739,7 +1738,6 @@ impl LanAddressPointer {
            .db
            .peek()
            .await
-            .map_err(|e| ConfigurationError::SystemError(e))?
            .as_package_data()
            .as_idx(&self.package_id)
            .and_then(|pde| pde.as_installed())
@@ -1775,11 +1773,7 @@ impl ConfigPointer {
            Ok(self.select(&Value::Object(cfg.clone())))
        } else {
            let id = &self.package_id;
-            let db = ctx
-                .db
-                .peek()
-                .await
-                .map_err(|e| ConfigurationError::SystemError(e))?;
+            let db = ctx.db.peek().await;
            let manifest = db.as_package_data().as_idx(id).map(|pde| pde.as_manifest());
            let cfg_actions = manifest.and_then(|m| m.as_config().transpose_ref());
            if let (Some(manifest), Some(cfg_actions)) = (manifest, cfg_actions) {
--- a/backend/src/context/rpc.rs
+++ b/backend/src/context/rpc.rs
@@ -276,7 +276,7 @@ impl RpcContext {
            })
            .await?;

-        let peek = self.db.peek().await?;
+        let peek = self.db.peek().await;

        for (package_id, package) in peek.as_package_data().as_entries()?.into_iter() {
            let action = match package.as_match() {
--- a/backend/src/control.rs
+++ b/backend/src/control.rs
@@ -12,7 +12,7 @@ use crate::Error;
 #[command(display(display_none), metadata(sync_db = true))]
 #[instrument(skip_all)]
 pub async fn start(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<(), Error> {
-    let peek = ctx.db.peek().await?;
+    let peek = ctx.db.peek().await;
    let version = peek
        .as_package_data()
        .as_idx(&id)
@@ -35,7 +35,7 @@ pub async fn start(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<(

 #[command(display(display_none), metadata(sync_db = true))]
 pub async fn stop(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<MainStatus, Error> {
-    let peek = ctx.db.peek().await?;
+    let peek = ctx.db.peek().await;
    let version = peek
        .as_package_data()
        .as_idx(&id)
@@ -71,7 +71,7 @@ pub async fn stop(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<Ma

 #[command(display(display_none), metadata(sync_db = true))]
 pub async fn restart(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<(), Error> {
-    let peek = ctx.db.peek().await?;
+    let peek = ctx.db.peek().await;
    let version = peek
        .as_package_data()
        .as_idx(&id)
--- a/backend/src/db/mod.rs
+++ b/backend/src/db/mod.rs
@@ -37,7 +37,7 @@ async fn ws_handler<
    session: Option<(HasValidSession, HashSessionToken)>,
    ws_fut: WSFut,
 ) -> Result<(), Error> {
-    let (dump, sub) = ctx.db.dump_and_sub().await?;
+    let (dump, sub) = ctx.db.dump_and_sub().await;
    let mut stream = ws_fut
        .await
        .with_kind(ErrorKind::Network)?
@@ -174,7 +174,7 @@ pub async fn subscribe(ctx: RpcContext, req: Request<Body>) -> Result<Response<B
    Ok(res)
 }

-#[command(subcommands(revisions, dump, put, apply))]
+#[command(subcommands(dump, put, apply))]
 pub fn db() -> Result<(), RpcError> {
    Ok(())
 }
@@ -186,20 +186,6 @@ pub enum RevisionsRes {
    Dump(Dump),
 }

-#[command(display(display_serializable))]
-pub async fn revisions(
-    #[context] ctx: RpcContext,
-    #[arg] since: u64,
-    #[allow(unused_variables)]
-    #[arg(long = "format")]
-    format: Option<IoFormat>,
-) -> Result<RevisionsRes, Error> {
-    Ok(match ctx.db.sync(since).await? {
-        Ok(revs) => RevisionsRes::Revisions(revs),
-        Err(dump) => RevisionsRes::Dump(dump),
-    })
-}
-
 #[instrument(skip_all)]
 async fn cli_dump(
    ctx: CliContext,
@@ -207,7 +193,7 @@ async fn cli_dump(
    path: Option<PathBuf>,
 ) -> Result<Dump, RpcError> {
    let dump = if let Some(path) = path {
-        PatchDb::open(path).await?.dump().await?
+        PatchDb::open(path).await?.dump().await
    } else {
        rpc_toolkit::command_helpers::call_remote(
            ctx,
@@ -235,7 +221,7 @@ pub async fn dump(
    #[arg]
    path: Option<PathBuf>,
 ) -> Result<Dump, Error> {
-    Ok(ctx.db.dump().await?)
+    Ok(ctx.db.dump().await)
 }

 fn apply_expr(input: jaq_core::Val, expr: &str) -> Result<jaq_core::Val, Error> {
--- a/backend/src/db/prelude.rs
+++ b/backend/src/db/prelude.rs
@@ -28,7 +28,7 @@ where

 #[async_trait::async_trait]
 pub trait PatchDbExt {
-    async fn peek(&self) -> Result<DatabaseModel, Error>;
+    async fn peek(&self) -> DatabaseModel;
    async fn mutate<U: UnwindSafe + Send>(
        &self,
        f: impl FnOnce(&mut DatabaseModel) -> Result<U, Error> + UnwindSafe + Send,
@@ -40,8 +40,8 @@ pub trait PatchDbExt {
 }
 #[async_trait::async_trait]
 impl PatchDbExt for PatchDb {
-    async fn peek(&self) -> Result<DatabaseModel, Error> {
-        Ok(DatabaseModel::from(self.dump().await?.value))
+    async fn peek(&self) -> DatabaseModel {
+        DatabaseModel::from(self.dump().await.value)
    }
    async fn mutate<U: UnwindSafe + Send>(
        &self,
--- a/backend/src/dependencies.rs
+++ b/backend/src/dependencies.rs
@@ -170,7 +170,7 @@ pub async fn configure_logic(
    ctx: RpcContext,
    (pkg_id, dependency_id): (PackageId, PackageId),
 ) -> Result<ConfigDryRes, Error> {
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;
    let pkg = db
        .as_package_data()
        .as_idx(&pkg_id)
--- a/backend/src/init.rs
+++ b/backend/src/init.rs
@@ -204,7 +204,7 @@ pub async fn init(cfg: &RpcContextConfig) -> Result<InitResult, Error> {
    let account = AccountInfo::load(&secret_store).await?;
    let db = cfg.db(&account).await?;
    tracing::info!("Opened PatchDB");
-    let peek = db.peek().await?;
+    let peek = db.peek().await;
    let mut server_info = peek.as_server_info().de()?;

    // write to ca cert store
--- a/backend/src/install/cleanup.rs
+++ b/backend/src/install/cleanup.rs
@@ -62,7 +62,7 @@ pub async fn cleanup_failed(ctx: &RpcContext, id: &PackageId) -> Result<(), Erro
    if let Some(version) = match ctx
        .db
        .peek()
-        .await?
+        .await
        .as_package_data()
        .as_idx(id)
        .or_not_found(id)?
@@ -141,7 +141,7 @@ pub async fn uninstall<Ex>(ctx: &RpcContext, secrets: &mut Ex, id: &PackageId) -
 where
    for<'a> &'a mut Ex: Executor<'a, Database = Postgres>,
 {
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;
    let entry = db
        .as_package_data()
        .as_idx(id)
--- a/backend/src/install/mod.rs
+++ b/backend/src/install/mod.rs
@@ -64,7 +64,7 @@ pub const PKG_WASM_DIR: &str = "package-data/wasm";

 #[command(display(display_serializable))]
 pub async fn list(#[context] ctx: RpcContext) -> Result<Value, Error> {
-    Ok(ctx.db.peek().await?.as_package_data().as_entries()?
+    Ok(ctx.db.peek().await.as_package_data().as_entries()?
        .iter()
        .filter_map(|(id, pde)| {
            let status = match pde.as_match() {
@@ -666,7 +666,7 @@ pub async fn download_install_s9pk(
 ) -> Result<(), Error> {
    let pkg_id = &temp_manifest.id;
    let version = &temp_manifest.version;
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;

    if let Result::<(), Error>::Err(e) = {
        let ctx = ctx.clone();
@@ -786,7 +786,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin + Send + Sync>(
    rdr.validated();
    let developer_key = rdr.developer_key().clone();
    rdr.reset().await?;
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;

    tracing::info!("Install {}@{}: Unpacking Manifest", pkg_id, version);
    let manifest = progress
@@ -1017,7 +1017,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin + Send + Sync>(
        )
        .await?;

-    let peek = ctx.db.peek().await?;
+    let peek = ctx.db.peek().await;
    let prev = peek
        .as_package_data()
        .as_idx(pkg_id)
--- a/backend/src/manager/health.rs
+++ b/backend/src/manager/health.rs
@@ -11,7 +11,7 @@ use crate::Error;
 #[instrument(skip_all)]
 pub async fn check(ctx: &RpcContext, id: &PackageId) -> Result<(), Error> {
    let (manifest, started) = {
-        let peeked = ctx.db.peek().await?;
+        let peeked = ctx.db.peek().await;
        let pde = peeked
            .as_package_data()
            .as_idx(id)
--- a/backend/src/manager/manager_container.rs
+++ b/backend/src/manager/manager_container.rs
@@ -53,7 +53,7 @@ impl ManageContainer {
        let current_state = Arc::new(watch::channel(StartStop::Stop).0);
        let desired_state = Arc::new(
            watch::channel::<StartStop>(
-                get_status(seed.ctx.db.peek().await?, &seed.manifest).into(),
+                get_status(seed.ctx.db.peek().await, &seed.manifest).into(),
            )
            .0,
        );
@@ -103,7 +103,7 @@ impl ManageContainer {
        &self,
        seed: &manager_seed::ManagerSeed,
    ) -> Result<(), Error> {
-        let current_state = get_status(seed.ctx.db.peek().await?, &seed.manifest);
+        let current_state = get_status(seed.ctx.db.peek().await, &seed.manifest);
        self.override_main_status
            .send_modify(|x| *x = Some(current_state));
        Ok(())
--- a/backend/src/manager/mod.rs
+++ b/backend/src/manager/mod.rs
@@ -267,7 +267,7 @@ impl Manager {
        let manage_container = self.manage_container.clone();
        let seed = self.seed.clone();
        async move {
-            let peek = seed.ctx.db.peek().await?;
+            let peek = seed.ctx.db.peek().await;
            let state_reverter = DesiredStateReverter::new(manage_container.clone());
            let override_guard =
                manage_container.set_override(get_status(peek, &seed.manifest).backing_up())?;
@@ -338,7 +338,7 @@ async fn configure(
    id: PackageId,
    mut configure_context: ConfigureContext,
 ) -> Result<BTreeMap<PackageId, String>, Error> {
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;
    let id = &id;
    let ctx = &ctx;
    let overrides = &mut configure_context.overrides;
--- a/backend/src/middleware/db.rs
+++ b/backend/src/middleware/db.rs
@@ -1,4 +1,3 @@
-use color_eyre::eyre::eyre;
 use futures::future::BoxFuture;
 use futures::FutureExt;
 use http::HeaderValue;
@@ -11,7 +10,6 @@ use rpc_toolkit::yajrc::RpcMethod;
 use rpc_toolkit::Metadata;

 use crate::context::RpcContext;
-use crate::{Error, ResultExt};

 pub fn db<M: Metadata>(ctx: RpcContext) -> DynMiddleware<M> {
    Box::new(
@@ -22,49 +20,19 @@ pub fn db<M: Metadata>(ctx: RpcContext) -> DynMiddleware<M> {
            async move {
                let m2: DynMiddlewareStage2 = Box::new(move |req, rpc_req| {
                    async move {
-                        let seq = req.headers.remove("x-patch-sequence");
                        let sync_db = metadata
                            .get(rpc_req.method.as_str(), "sync_db")
                            .unwrap_or(false);

                        let m3: DynMiddlewareStage3 = Box::new(move |res, _| {
                            async move {
-                                if sync_db && seq.is_some() {
-                                    match async {
-                                        let seq = seq
-                                            .ok_or_else(|| {
-                                                Error::new(
-                                                    eyre!("Missing X-Patch-Sequence"),
-                                                    crate::ErrorKind::InvalidRequest,
-                                                )
-                                            })?
-                                            .to_str()
-                                            .with_kind(crate::ErrorKind::InvalidRequest)?
-                                            .parse()?;
-                                        let res = ctx.db.sync(seq).await?;
-                                        let json = match res {
-                                            Ok(revs) => serde_json::to_vec(&revs),
-                                            Err(dump) => serde_json::to_vec(&[dump]),
-                                        }
-                                        .with_kind(crate::ErrorKind::Serialization)?;
-                                        Ok::<_, Error>(base64::encode_config(
-                                            &json,
-                                            base64::URL_SAFE,
-                                        ))
-                                    }
-                                    .await
-                                    {
-                                        Ok(a) => res
-                                            .headers
-                                            .append("X-Patch-Updates", HeaderValue::from_str(&a)?),
-                                        Err(e) => res.headers.append(
-                                            "X-Patch-Error",
-                                            HeaderValue::from_str(&base64::encode_config(
-                                                &e.to_string(),
-                                                base64::URL_SAFE,
-                                            ))?,
-                                        ),
-                                    };
+                                if sync_db {
+                                    res.headers.append(
+                                        "X-Patch-Sequence",
+                                        HeaderValue::from_str(
+                                            &ctx.db.sequence().await.to_string(),
+                                        )?,
+                                    );
                                }
                                Ok(Ok(noop4()))
                            }
--- a/backend/src/notifications.rs
+++ b/backend/src/notifications.rs
@@ -236,7 +236,7 @@ impl NotificationManager {
        subtype: T,
        debounce_interval: Option<u32>,
    ) -> Result<(), Error> {
-        let peek = db.peek().await?;
+        let peek = db.peek().await;
        if !self
            .should_notify(&package_id, &level, &title, debounce_interval)
            .await
--- a/backend/src/properties.rs
+++ b/backend/src/properties.rs
@@ -21,7 +21,7 @@ pub async fn properties(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Res

 #[instrument(skip_all)]
 pub async fn fetch_properties(ctx: RpcContext, id: PackageId) -> Result<Value, Error> {
-    let peek = ctx.db.peek().await?;
+    let peek = ctx.db.peek().await;

    let manifest = peek
        .as_package_data()
--- a/backend/src/system.rs
+++ b/backend/src/system.rs
@@ -58,7 +58,7 @@ pub async fn enable_zram() -> Result<(), Error> {

 #[command(display(display_none))]
 pub async fn zram(#[context] ctx: RpcContext, #[arg] enable: bool) -> Result<(), Error> {
-    let db = ctx.db.peek().await?;
+    let db = ctx.db.peek().await;

    let zram = db.as_server_info().as_zram().de()?;
    if enable == zram {
--- a/backend/src/update/mod.rs
+++ b/backend/src/update/mod.rs
@@ -76,7 +76,7 @@ fn display_update_result(status: UpdateResult, _: &ArgMatches) {

 #[instrument(skip_all)]
 async fn maybe_do_update(ctx: RpcContext, marketplace_url: Url) -> Result<Option<()>, Error> {
-    let peeked = ctx.db.peek().await?;
+    let peeked = ctx.db.peek().await;
    let latest_version: Version = ctx
        .client
        .get(with_query_params(
--- a/backend/src/version/mod.rs
+++ b/backend/src/version/mod.rs
@@ -163,7 +163,7 @@ where
 }

 pub async fn init(db: &PatchDb, secrets: &PgPool) -> Result<(), Error> {
-    let version = Version::from_util_version(db.peek().await?.as_server_info().as_version().de()?);
+    let version = Version::from_util_version(db.peek().await.as_server_info().as_version().de()?);

    match version {
        Version::V0_3_4(v) => v.0.migrate_to(&Current::new(), db.clone(), secrets).await?,
--- a/backend/src/version/v0_3_5.rs
+++ b/backend/src/version/v0_3_5.rs
@@ -28,7 +28,7 @@ impl VersionT for Version {
        &V0_3_0_COMPAT
    }
    async fn up(&self, db: PatchDb, _secrets: &PgPool) -> Result<(), Error> {
-        let peek = db.peek().await?;
+        let peek = db.peek().await;
        let mut url_replacements = BTreeMap::new();
        for (_, pde) in peek.as_package_data().as_entries()? {
            for (dependency, info) in pde