Add sync health and explicit sync trigger

crates/geth-control/src/lib.rs

@@ -189,6 +189,10 @@ pub enum ControlRequest {
     AuthSync {
         node: String,
     },
+    SyncStatus,
+    SyncNow {
+        node: Option<String>,
+    },
     SecretStatus,
     SecretCreate {
         resource: String,
@@ -576,6 +580,7 @@ pub enum ControlResponse {
         ops_imported: usize,
         signatures_imported: usize,
         invalid_ops_rejected: usize,
+        high_water_ms: i64,
         note: String,
     },
     AuthSynced {
@@ -585,6 +590,15 @@ pub enum ControlResponse {
         ops_imported: usize,
         signatures_imported: usize,
         invalid_ops_rejected: usize,
+        high_water_ms: i64,
+        note: String,
+    },
+    SyncStatus {
+        peers: Vec<SyncPeerStatus>,
+        note: String,
+    },
+    SyncRan {
+        peers: Vec<SyncPeerRun>,
         note: String,
     },
     SecretStatus {
@@ -953,10 +967,12 @@ pub enum PeerControlRequest {
     },
     KeychainSync {
         peer_card: PeerCard,
+        since_ms: i64,
         nonce: String,
     },
     AuthSync {
         peer_card: PeerCard,
+        since_ms: i64,
         nonce: String,
     },
     NodeEnrollmentSubmit {
@@ -1089,6 +1105,7 @@ pub enum PeerControlResponse {
         remote_endpoint_id: String,
         ops: Vec<KeychainOp>,
         signatures: Vec<KeychainOpSignature>,
+        high_water_ms: i64,
         nonce: String,
         note: String,
     },
@@ -1099,6 +1116,7 @@ pub enum PeerControlResponse {
         remote_endpoint_id: String,
         ops: Vec<AuthOp>,
         signatures: Vec<AuthOpSignature>,
+        high_water_ms: i64,
         nonce: String,
         note: String,
     },
@@ -1351,6 +1369,40 @@ pub struct SyncWatermark {
     pub high_water: i64,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SyncPeerStatus {
+    pub peer_node_id: String,
+    pub streams: Vec<SyncStreamStatus>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SyncStreamStatus {
+    pub stream: String,
+    pub cursor_ms: i64,
+    pub last_attempt_ms: Option<i64>,
+    pub last_success_ms: Option<i64>,
+    pub last_error: Option<String>,
+    pub last_imported: usize,
+    pub last_rejected: usize,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SyncPeerRun {
+    pub peer_node_id: String,
+    pub streams: Vec<SyncStreamRun>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SyncStreamRun {
+    pub stream: String,
+    pub attempted: bool,
+    pub success: bool,
+    pub imported: usize,
+    pub rejected: usize,
+    pub cursor_ms: i64,
+    pub error: Option<String>,
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum ControlError {
     #[error("json error: {0}")]
@@ -1507,6 +1559,7 @@ mod tests {
             ops_imported: 2,
             signatures_imported: 2,
             invalid_ops_rejected: 1,
+            high_water_ms: 42,
             note: "trusted admin signatures only".to_owned(),
         };
         assert_eq!(
@@ -1535,6 +1588,7 @@ mod tests {
             ops_imported: 1,
             signatures_imported: 1,
             invalid_ops_rejected: 0,
+            high_water_ms: 42,
             note: "trusted admin signatures only".to_owned(),
         };
         assert_eq!(
@@ -1542,6 +1596,46 @@ mod tests {
             response
         );
 
+        let response = ControlResponse::SyncStatus {
+            peers: vec![SyncPeerStatus {
+                peer_node_id: "node:peer".to_owned(),
+                streams: vec![SyncStreamStatus {
+                    stream: "keychain".to_owned(),
+                    cursor_ms: 42,
+                    last_attempt_ms: Some(43),
+                    last_success_ms: Some(43),
+                    last_error: None,
+                    last_imported: 2,
+                    last_rejected: 0,
+                }],
+            }],
+            note: "local health".to_owned(),
+        };
+        assert_eq!(
+            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
+            response
+        );
+
+        let response = ControlResponse::SyncRan {
+            peers: vec![SyncPeerRun {
+                peer_node_id: "node:peer".to_owned(),
+                streams: vec![SyncStreamRun {
+                    stream: "auth".to_owned(),
+                    attempted: true,
+                    success: true,
+                    imported: 1,
+                    rejected: 0,
+                    cursor_ms: 44,
+                    error: None,
+                }],
+            }],
+            note: "ran".to_owned(),
+        };
+        assert_eq!(
+            decode_response(&encode_response(&response).expect("encode")).expect("decode"),
+            response
+        );
+
         let request = ControlRequest::SecretBearerVerify {
             secret: "bearer:test".to_owned(),
             resource: "resource:cas:local".to_owned(),