fix catch-up

1 files changed, 41 insertions, 4 deletions

diff --git a/src/mycelium.rs b/src/mycelium.rs
index 22d49c1..96b9e6d 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -576,6 +576,21 @@ impl Transport for MyceliumTransport {
             }
             .into());
         }
+
+        // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing
+        // a message that never arrived.
+        if tracing::enabled!(tracing::Level::DEBUG) {
+            if let Ok(body) = response
+                .bytes()
+                .await
+                .map_err(anyhow::Error::from)
+                .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?))
+            {
+                if let Some(id) = body.get("id").and_then(|id| id.as_str()) {
+                    tracing::debug!(peer = %message.peer, message_id = id, "Mycelium accepted message");
+                }
+            }
+        }
         Ok(())
     }
 
@@ -919,18 +934,25 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
 struct SnapshotRequestShadow {
     epoch: EpochShadow,
     wanted: EpochShadow,
+    offset: u64,
 }
 
 #[derive(Serialize, Deserialize)]
 struct SnapshotReplyShadow {
     epoch: EpochShadow,
-    document: ByteBuf,
+    wanted: EpochShadow,
+    digest: ByteBuf,
+    total: u64,
+    offset: u64,
+    data: ByteBuf,
+    refused: String,
 }
 
 fn snapshot_request_to_bytes(msg: &SnapshotRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&SnapshotRequestShadow {
         epoch: epoch_to_shadow(&msg.epoch),
         wanted: epoch_to_shadow(&msg.wanted),
+        offset: msg.offset,
     })
 }
 
@@ -939,22 +961,37 @@ fn bytes_to_snapshot_request(bytes: &[u8]) -> Result<SnapshotRequest, TransportE
     Ok(SnapshotRequest {
         epoch: epoch_from_shadow(shadow.epoch)?,
         wanted: epoch_from_shadow(shadow.wanted)?,
+        offset: shadow.offset,
     })
 }
 
 fn snapshot_reply_to_bytes(msg: &SnapshotReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&SnapshotReplyShadow {
         epoch: epoch_to_shadow(&msg.epoch),
-        // One copy of the document; replies are rare.
-        document: ByteBuf::from(msg.document.clone()),
+        wanted: epoch_to_shadow(&msg.wanted),
+        digest: ByteBuf::from(msg.digest.to_vec()),
+        total: msg.total,
+        offset: msg.offset,
+        data: ByteBuf::from(msg.data.clone()),
+        refused: msg.refused.clone(),
     })
 }
 
 fn bytes_to_snapshot_reply(bytes: &[u8]) -> Result<SnapshotReply, TransportError> {
     let shadow: SnapshotReplyShadow = rmp_serde::from_slice(bytes)?;
+    let digest = shadow
+        .digest
+        .as_slice()
+        .try_into()
+        .map_err(|_| TransportError::InvalidPayload("digest must be 32 bytes".into()))?;
     Ok(SnapshotReply {
         epoch: epoch_from_shadow(shadow.epoch)?,
-        document: shadow.document.into_vec(),
+        wanted: epoch_from_shadow(shadow.wanted)?,
+        digest,
+        total: shadow.total,
+        offset: shadow.offset,
+        data: shadow.data.into_vec(),
+        refused: shadow.refused,
     })
 }