Add lots of extra error contexts

Author: Quantumplation

src/archive.rs

@@ -103,7 +103,12 @@ impl Archive {
         let header = block.header.as_ref().expect("must have header");
 
         // Save the raw bytes of the block, indexed by its hash
-        self.save_raw_block(&header.hash, bytes).await?;
+        self.save_raw_block(&header.hash, bytes)
+            .await
+            .context(format!(
+                "failed to save raw block {}",
+                hex::encode(&header.hash)
+            ))?;
 
         // Then, save various lookups in dynamodb
         let location = block_hash_key(&header.hash);

src/broadcast/broadcast.rs

@@ -1,6 +1,6 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use aws_sdk_dynamodb::{types::AttributeValue, Client as DynamoClient};
 use aws_sdk_kinesis::Client as KinesisClient;
 use aws_sdk_s3::primitives::Blob;
@@ -85,7 +85,8 @@ impl Broadcaster {
                             .as_millis() as u64;
                         now < *deadline
                     })
-                    .await?;
+                    .await
+                    .context("failed checking for deadline")?;
 
                 // then send to kinesis, and save the point/seq number back to the destination
                 let result = self
@@ -95,15 +96,17 @@ impl Broadcaster {
                     .data(Blob::new(message_bytes.clone()))
                     .stream_arn(&destination.stream_arn)
                     .send()
-                    .await?;
+                    .await
+                    .context("failed writing to kinesis")?;
                 destination
                     .commit(
                         &self.dynamo,
                         &self.table,
                         message.advance.clone(),
                         Some(result.sequence_number),
                     )
-                    .await?;
+                    .await
+                    .context(format!("failed committing destination {}", destination.pk))?;
                 destinations.push(destination.pk.clone());
             } else {
                 // If the block doesn't apply, we still advance the point
@@ -117,7 +120,11 @@ impl Broadcaster {
                         message.advance.clone(),
                         destination.sequence_number.clone(),
                     )
-                    .await?;
+                    .await
+                    .context(format!(
+                        "failed advancing sequence number for destination {}",
+                        destination.pk
+                    ))?;
             }
         }
         Ok(destinations)

src/broadcast/destination.rs

@@ -1,7 +1,7 @@
 use crate::broadcast::BroadcastMessage;
 
 use super::filter::FilterConfig;
-use anyhow::Result;
+use anyhow::{Context, Result};
 use aws_sdk_dynamodb::{types::AttributeValue, Client as DynamoClient};
 use aws_sdk_kinesis::{types::ShardIteratorType, Client as KinesisClient};
 use bytes::Bytes;
@@ -75,7 +75,8 @@ impl Destination {
             .expression_attribute_values(":new_point", AttributeValue::S(point_to_string(&point)))
             .expression_attribute_values(":rotated_points", AttributeValue::L(recovery_points))
             .send()
-            .await?;
+            .await
+            .context(format!("failed to update destination {}", self.pk))?;
         Ok(())
     }
 
@@ -107,21 +108,26 @@ impl Destination {
         }
         let mut iterator = shard_request
             .send()
-            .await?
+            .await
+            .context("failed to request shard iterator")?
             .shard_iterator
-            .expect("failed to get shard iterator");
+            .context("shard iterator is none")?;
         info!("Repairing destination {}", self.pk);
         loop {
             let records = kinesis
                 .get_records()
                 .stream_arn(&self.stream_arn)
                 .shard_iterator(&iterator)
                 .send()
-                .await?;
+                .await
+                .context(format!(
+                    "failed to fetch records from stream {}",
+                    self.stream_arn
+                ))?;
 
             iterator = records
                 .next_shard_iterator
-                .expect("stream should be provisioned");
+                .context("next shard iterator is none")?;
 
             info!(
                 "Received {} records, {:?}ms behind tip",
@@ -133,9 +139,12 @@ impl Destination {
                 let last_record = records.records.into_iter().last().unwrap();
                 let seq_no = last_record.sequence_number;
                 let data = last_record.data.into_inner();
-                let message: BroadcastMessage = serde_json::from_slice(data.as_slice())?;
+                let message: BroadcastMessage =
+                    serde_json::from_slice(data.as_slice()).context("failed to parse data")?;
                 let advance = message.advance;
-                self.commit(&dynamo, &table, advance, Some(seq_no)).await?;
+                self.commit(&dynamo, &table, advance, Some(seq_no))
+                    .await
+                    .context("failed to commit while repairing")?;
             }
 
             if millis_behind_latest == 0 {

src/worker/worker.rs

@@ -13,7 +13,7 @@ use crate::{
     broadcast::{BroadcastMessage, Broadcaster},
     utils::elapsed,
 };
-use anyhow::{bail, Result};
+use anyhow::{bail, Context, Result};
 
 pub struct Worker {
     pub dynamo: DynamoClient,
@@ -36,14 +36,18 @@ impl Worker {
             self.kinesis.clone(),
             lock_deadline,
         )
-        .await?;
+        .await
+        .context("failed to start broadcaster")?;
 
         if broadcaster.destinations.is_empty() {
             warn!("No enabled destinations, nothing to do");
             return Ok(());
         }
 
-        broadcaster.repair().await?;
+        broadcaster
+            .repair()
+            .await
+            .context("failed to repair broadcaster")?;
 
         let earliest_point = broadcaster
             .destinations
@@ -66,7 +70,9 @@ impl Worker {
                 .to_vec()
                 .encode_hex::<String>()
         );
-        let mut follower = Follower::new(&self.uri, &self.api_key, intersect).await?;
+        let mut follower = Follower::new(&self.uri, &self.api_key, intersect)
+            .await
+            .context("failed to start follower")?;
 
         let mut undo_stack = vec![];
 
@@ -77,7 +83,7 @@ impl Worker {
                     bail!("No block in 5 minutes, failing over to another node");
                 },
                 result = follower.next_event() => {
-                    let (is_roll_forward, bytes, block, header) = result?;
+                    let (is_roll_forward, bytes, block, header) = result.context("failed to receive next event from follower")?;
                     let block_hash: String = header.hash.encode_hex();
 
                     let start = SystemTime::now();
@@ -86,19 +92,19 @@ impl Worker {
                         hash: header.hash,
                     };
                     if is_roll_forward {
-                        self.archive.save(&block, bytes.to_vec()).await?;
+                        self.archive.save(&block, bytes.to_vec()).await.context(format!("failed to archive {}/{}", point.index, block_hash))?;
 
                         let start = SystemTime::now();
                         let destinations = broadcaster.broadcast(block, BroadcastMessage {
                             undo: undo_stack.clone(),
                             advance: point.clone(),
-                        }).await?;
+                        }).await.context(format!("failed to broadcast point {}/{}", point.index, block_hash))?;
                         trace!("Message broadcast (elapsed={:?})", SystemTime::now().duration_since(start)?);
                         undo_stack.clear();
                         info!("Roll forward {}/{} ({})", point.index, block_hash, destinations.join(", "));
                     } else {
                         trace!("Unsaving {}/{}", point.index, block_hash);
-                        self.archive.unsave(&block).await?;
+                        self.archive.unsave(&block).await.context(format!("failed to unsave point {}/{}", point.index, block_hash))?;
                         trace!("Block {}/{} unsaved (elapsed={:?})", point.index, block_hash, elapsed(start));
                         undo_stack.push(point.clone());
                         info!("Undo block {}/{}", point.index, block_hash);