diff --git a/bin/core/src/api/execute/sync.rs b/bin/core/src/api/execute/sync.rs index efdd2d634..3c9952bba 100644 --- a/bin/core/src/api/execute/sync.rs +++ b/bin/core/src/api/execute/sync.rs @@ -18,7 +18,7 @@ use monitor_client::{ user::{sync_user, User}, }, }; -use mungos::by_id::update_one_by_id; +use mungos::{by_id::update_one_by_id, mongodb::bson::to_document}; use resolver_api::Resolve; use serror::serialize_error_pretty; @@ -33,7 +33,7 @@ use crate::{ }, update::update_update, }, - resource, + resource::{self, refresh_resource_sync_state_cache}, state::{db_client, State}, }; @@ -319,8 +319,10 @@ impl Resolve for State { .await, ); + let db = db_client().await; + if let Err(e) = update_one_by_id( - &db_client().await.resource_syncs, + &db.resource_syncs, &sync.id, doc! { "$set": { @@ -357,6 +359,21 @@ impl Resolve for State { } update.finalize(); + + // Need to manually update the update before cache refresh, + // and before broadcast with add_update. + // The Err case of to_document should be unreachable, + // but will fail to update cache in that case. + if let Ok(update_doc) = to_document(&update) { + let _ = update_one_by_id( + &db.updates, + &update.id, + mungos::update::Update::Set(update_doc), + None, + ) + .await; + refresh_resource_sync_state_cache().await; + } update_update(update.clone()).await?; Ok(update) diff --git a/bin/core/src/api/read/sync.rs b/bin/core/src/api/read/sync.rs index 773abb1d7..006b1030f 100644 --- a/bin/core/src/api/read/sync.rs +++ b/bin/core/src/api/read/sync.rs @@ -4,8 +4,8 @@ use monitor_client::{ entities::{ permission::PermissionLevel, sync::{ - ResourceSync, ResourceSyncActionState, ResourceSyncListItem, - ResourceSyncState, + PendingSyncUpdatesData, ResourceSync, ResourceSyncActionState, + ResourceSyncListItem, ResourceSyncState, }, user::User, }, @@ -96,6 +96,19 @@ impl Resolve for State { for resource_sync in resource_syncs { res.total += 1; + match resource_sync.info.pending.data { + PendingSyncUpdatesData::Ok(data) => { + if !data.no_updates() { + res.pending += 1; + continue; + } + } + PendingSyncUpdatesData::Err(_) => { + res.failed += 1; + continue; + } + } + match ( cache.get(&resource_sync.id).await.unwrap_or_default(), action_states @@ -115,6 +128,9 @@ impl Resolve for State { (ResourceSyncState::Syncing, _) => { unreachable!() } + (ResourceSyncState::Pending, _) => { + unreachable!() + } } } diff --git a/bin/core/src/main.rs b/bin/core/src/main.rs index 0d90f9116..b44b64ff7 100644 --- a/bin/core/src/main.rs +++ b/bin/core/src/main.rs @@ -38,6 +38,7 @@ async fn app() -> anyhow::Result<()> { resource::spawn_build_state_refresh_loop(); resource::spawn_repo_state_refresh_loop(); resource::spawn_procedure_state_refresh_loop(); + resource::spawn_resource_sync_state_refresh_loop(); // Setup static frontend services let frontend_path = frontend_path(); diff --git a/bin/core/src/resource/mod.rs b/bin/core/src/resource/mod.rs index 9f01944d6..be8bf11db 100644 --- a/bin/core/src/resource/mod.rs +++ b/bin/core/src/resource/mod.rs @@ -59,6 +59,10 @@ pub use procedure::{ pub use repo::{ refresh_repo_state_cache, spawn_repo_state_refresh_loop, }; +pub use sync::{ + refresh_resource_sync_state_cache, + spawn_resource_sync_state_refresh_loop, +}; /// Implement on each monitor resource for common methods pub trait MonitorResource { diff --git a/bin/core/src/resource/sync.rs b/bin/core/src/resource/sync.rs index bae085614..d62c2c588 100644 --- a/bin/core/src/resource/sync.rs +++ b/bin/core/src/resource/sync.rs @@ -1,18 +1,27 @@ +use std::time::Duration; + +use anyhow::Context; +use mongo_indexed::doc; use monitor_client::entities::{ resource::Resource, sync::{ - PartialResourceSyncConfig, ResourceSync, ResourceSyncConfig, - ResourceSyncConfigDiff, ResourceSyncInfo, ResourceSyncListItem, - ResourceSyncListItemInfo, ResourceSyncQuerySpecifics, - ResourceSyncState, + PartialResourceSyncConfig, PendingSyncUpdatesData, ResourceSync, + ResourceSyncConfig, ResourceSyncConfigDiff, ResourceSyncInfo, + ResourceSyncListItem, ResourceSyncListItemInfo, + ResourceSyncQuerySpecifics, ResourceSyncState, }, update::{ResourceTargetVariant, Update}, user::User, Operation, }; -use mungos::mongodb::Collection; +use mungos::{ + find::find_collect, + mongodb::{options::FindOneOptions, Collection}, +}; -use crate::state::{action_states, db_client}; +use crate::state::{ + action_states, db_client, resource_sync_state_cache, +}; impl super::MonitorResource for ResourceSync { type Config = ResourceSyncConfig; @@ -34,6 +43,11 @@ impl super::MonitorResource for ResourceSync { async fn to_list_item( resource_sync: Resource, ) -> Self::ListItem { + let state = get_resource_sync_state( + &resource_sync.id, + &resource_sync.info.pending.data, + ) + .await; ResourceSyncListItem { id: resource_sync.id, name: resource_sync.name, @@ -42,10 +56,10 @@ impl super::MonitorResource for ResourceSync { info: ResourceSyncListItemInfo { repo: resource_sync.config.repo, branch: resource_sync.config.branch, - last_sync_ts: 0, - last_sync_hash: String::new(), - last_sync_message: String::new(), - state: ResourceSyncState::Unknown, + last_sync_ts: resource_sync.info.last_sync_ts, + last_sync_hash: resource_sync.info.last_sync_hash, + last_sync_message: resource_sync.info.last_sync_message, + state, }, } } @@ -124,3 +138,107 @@ impl super::MonitorResource for ResourceSync { Ok(()) } } + +pub fn spawn_resource_sync_state_refresh_loop() { + tokio::spawn(async move { + loop { + refresh_resource_sync_state_cache().await; + tokio::time::sleep(Duration::from_secs(60)).await; + } + }); +} + +pub async fn refresh_resource_sync_state_cache() { + let _ = async { + let resource_syncs = + find_collect(&db_client().await.resource_syncs, None, None) + .await + .context("failed to get resource_syncs from db")?; + let cache = resource_sync_state_cache(); + for resource_sync in resource_syncs { + let state = + get_resource_sync_state_from_db(&resource_sync.id).await; + cache.insert(resource_sync.id, state).await; + } + anyhow::Ok(()) + } + .await + .inspect_err(|e| { + error!("failed to refresh resource_sync state cache | {e:#}") + }); +} + +async fn get_resource_sync_state( + id: &String, + data: &PendingSyncUpdatesData, +) -> ResourceSyncState { + if let Some(state) = action_states() + .resource_sync + .get(id) + .await + .and_then(|s| { + s.get() + .map(|s| { + if s.syncing { + Some(ResourceSyncState::Syncing) + } else { + None + } + }) + .ok() + }) + .flatten() + { + return state; + } + let data = match data { + PendingSyncUpdatesData::Err(_) => { + return ResourceSyncState::Failed + } + PendingSyncUpdatesData::Ok(data) => data, + }; + if !data.no_updates() { + return ResourceSyncState::Pending; + } + resource_sync_state_cache() + .get(id) + .await + .unwrap_or_default() +} + +async fn get_resource_sync_state_from_db( + id: &str, +) -> ResourceSyncState { + async { + let state = db_client() + .await + .updates + .find_one( + doc! { + "target.type": "ResourceSync", + "target.id": id, + "operation": "RunSync" + }, + FindOneOptions::builder() + .sort(doc! { "start_ts": -1 }) + .build(), + ) + .await? + .map(|u| { + if u.success { + ResourceSyncState::Ok + } else { + ResourceSyncState::Failed + } + }) + .unwrap_or(ResourceSyncState::Ok); + anyhow::Ok(state) + } + .await + .inspect_err(|e| { + warn!( + "failed to get resource sync state from db for {id} | {e:#}" + ) + }) + .unwrap_or(ResourceSyncState::Unknown) +} diff --git a/client/core/rs/src/api/read/sync.rs b/client/core/rs/src/api/read/sync.rs index c15deaf07..de67757d7 100644 --- a/client/core/rs/src/api/read/sync.rs +++ b/client/core/rs/src/api/read/sync.rs @@ -104,6 +104,8 @@ pub struct GetResourceSyncsSummaryResponse { pub ok: u32, /// The number of syncs currently syncing. pub syncing: u32, + /// The number of syncs with pending updates + pub pending: u32, /// The number of syncs with failed state. pub failed: u32, /// The number of syncs with unknown state. diff --git a/client/core/rs/src/entities/sync.rs b/client/core/rs/src/entities/sync.rs index 8d8a05fa7..2e8cadcde 100644 --- a/client/core/rs/src/entities/sync.rs +++ b/client/core/rs/src/entities/sync.rs @@ -37,12 +37,14 @@ pub struct ResourceSyncListItemInfo { Debug, Clone, Copy, Default, Serialize, Deserialize, Display, )] pub enum ResourceSyncState { - /// Last sync successful (or never synced) + /// Last sync successful (or never synced). No Changes pending Ok, /// Last sync failed Failed, /// Currently syncing Syncing, + /// Updates pending + Pending, /// Other case #[default] Unknown, diff --git a/client/core/ts/src/types.ts b/client/core/ts/src/types.ts index 7fc048006..bf2e89ace 100644 --- a/client/core/ts/src/types.ts +++ b/client/core/ts/src/types.ts @@ -1210,12 +1210,14 @@ export type ResourceSync = Resource; export type GetResourceSyncResponse = ResourceSync; export enum ResourceSyncState { - /** Last sync successful (or never synced) */ + /** Last sync successful (or never synced). No Changes pending */ Ok = "Ok", /** Last sync failed */ Failed = "Failed", /** Currently syncing */ Syncing = "Syncing", + /** Updates pending */ + Pending = "Pending", /** Other case */ Unknown = "Unknown", } @@ -2656,6 +2658,8 @@ export interface GetResourceSyncsSummaryResponse { ok: number; /** The number of syncs currently syncing. */ syncing: number; + /** The number of syncs with pending updates */ + pending: number; /** The number of syncs with failed state. */ failed: number; /** The number of syncs with unknown state. */