1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
//! Compaction: replacing the document with a snapshot in a new epoch.
//!
//! * `snapshot`: deterministic snapshot build (step 3).
//! * `records`: the replicated compaction records under `_continuum/compactions/<id>`.
//! * `coordinator`: proposal, settle period and decision on the node that was asked.
//! * `switch`: switching this node to a committed epoch (worker).
//! * `delta`: replaying this node's changes since the base into the new epoch.
//! * `local`: per-node in-memory state (prebuilt snapshots, switch status, archive).
//! * `api`: `/admin/compaction(s)` handlers.
//!
//! The rest of the node is reached exclusively through [`Host`], so this module never
//! depends on `AppState`, peers or sync internals.
mod api;
mod coordinator;
mod delta;
mod local;
mod records;
pub(crate) mod snapshot;
mod switch;
pub(crate) use local::Compaction;
pub(crate) use switch::run_worker;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
use automerge::AutoCommit;
use axum::{
routing::{get, post},
Router,
};
use tokio::sync::broadcast;
use crate::replica::Epoch;
use crate::sync::ChangeSet;
/// What compaction needs from the node. Implemented by `AppState`.
///
/// Used generically (`H: Host`), never as `dyn Host`, so the generic methods are fine.
pub(crate) trait Host: Clone + Send + Sync + 'static {
/// This node's in-memory compaction state.
fn compaction(&self) -> &Compaction;
/// This node's peer ID.
fn node_id(&self) -> String;
/// The epoch of the local replica.
fn current_epoch(&self) -> Epoch;
/// Lower bound and default basis for the settle period.
fn dormant_after(&self) -> Duration;
fn is_node_ready(&self) -> impl Future<Output = bool> + Send;
/// Known (not forgotten) peers.
fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send;
/// Store paths touched by each published document change.
fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>>;
/// Reads the document on the blocking pool.
fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
where
F: FnOnce(&AutoCommit) -> T + Send + 'static,
T: Send + 'static;
/// Mutable access to the document without publishing a change (e.g. forks).
fn update_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
where
F: FnOnce(&mut AutoCommit) -> T + Send + 'static,
T: Send + 'static;
/// One committed change (rolled back on error), published and replicated.
fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
where
F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
T: Send + 'static,
E: Send + 'static;
/// Replaces the document and its epoch atomically. Under the write lock, `f` gets
/// the live document and returns the new one. Afterwards the node resets every
/// peer's sync state and publishes the change. Returns the old document.
fn switch_epoch<E, F>(
&self,
epoch: Epoch,
f: F,
) -> impl Future<Output = Result<AutoCommit, E>> + Send
where
F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static,
E: Send + 'static;
}
#[derive(Clone, Debug)]
pub(crate) struct PeerSummary {
pub(crate) id: String,
/// Not dormant.
pub(crate) active: bool,
}
/// Admin routes. Merged into the authenticated part of the node's router.
pub(crate) fn router<H: Host>() -> Router<H> {
Router::new()
.route(
"/admin/compaction",
post(api::start::<H>).delete(api::abort_current::<H>),
)
.route("/admin/compactions", get(api::list::<H>))
.route(
"/admin/compactions/{id}",
get(api::get::<H>).delete(api::abort::<H>),
)
.route("/admin/epochs", get(api::epochs::<H>))
}
|
