jcoffey-dev
0755fad51e
A FETCH with CHANGEDSINCE presents its mod-sequence to the read scope, so a replica must have that change before it answers, as a JMAP sinceState already did. replica_cluster_tests covers test 11: with the replica's replay paused, a write on one node is read back through a second store with its own marks, sharing through Redis. Composite stores nest store futures deeply enough to pass rustc's default query depth once both postgres and redis are compiled in, so the server crates raise their recursion limit.
71 lines
1.9 KiB
Rust
71 lines
1.9 KiB
Rust
/*
|
|
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
|
*
|
|
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
|
*/
|
|
|
|
// inbuxa: composite stores (sharded members, read replicas) nest store
|
|
// futures deeply enough to pass rustc's default query depth
|
|
#![recursion_limit = "512"]
|
|
|
|
#![warn(clippy::large_futures)]
|
|
|
|
use broadcast::publisher::spawn_broadcast_publisher;
|
|
use common::{
|
|
BuildServer, Inner,
|
|
manager::boot::{BootManager, IpcReceivers},
|
|
};
|
|
use state_manager::manager::spawn_push_router;
|
|
use std::sync::Arc;
|
|
|
|
use crate::task_manager::{manager::spawn_task_manager, scheduler::spawn_task_scheduler};
|
|
|
|
pub mod broadcast;
|
|
pub mod state_manager;
|
|
pub mod task_manager;
|
|
|
|
pub trait StartServices: Sync + Send {
|
|
fn start_services(&mut self) -> impl Future<Output = ()> + Send;
|
|
}
|
|
|
|
pub trait SpawnServices {
|
|
fn spawn_services(&mut self, inner: Arc<Inner>);
|
|
}
|
|
|
|
impl StartServices for BootManager {
|
|
async fn start_services(&mut self) {
|
|
let server = self.inner.build_server();
|
|
// Unpack webadmin
|
|
self.inner
|
|
.data
|
|
.applications
|
|
.unpack_all(&server, false)
|
|
.await;
|
|
|
|
if !server.registry().is_recovery_mode() {
|
|
self.ipc_rxs.spawn_services(self.inner.clone());
|
|
}
|
|
}
|
|
}
|
|
|
|
impl SpawnServices for IpcReceivers {
|
|
fn spawn_services(&mut self, inner: Arc<Inner>) {
|
|
if !inner.shared_core.load().storage.registry.is_recovery_mode() {
|
|
// Spawn push manager
|
|
spawn_push_router(inner.clone(), self.push_rx.take().unwrap());
|
|
|
|
// Spawn broadcast publisher
|
|
if let Some(event_rx) = self.broadcast_rx.take() {
|
|
// Spawn broadcast publisher
|
|
spawn_broadcast_publisher(inner.clone(), event_rx);
|
|
}
|
|
|
|
// Spawn task manager
|
|
spawn_task_manager(inner.clone());
|
|
|
|
// Spawn task scheduler
|
|
spawn_task_scheduler(inner);
|
|
}
|
|
}
|
|
}
|