Less logging (#81)

* reduce logs by a ton

* don't try to send a request if there was no statements to send

* relax 'no good candidates' errors

Author: jeromegn

crates/corro-agent/src/agent.rs

@@ -957,7 +957,7 @@ async fn clear_overwritten_versions(agent: Agent) {
             interval = COMPACT_BOOKED_INTERVAL;
         }
 
-        info!("starting compaction...");
+        info!("Starting compaction...");
         let start = Instant::now();
 
         let mut to_check: BTreeMap<i64, (ActorId, i64)> = BTreeMap::new();
@@ -991,7 +991,7 @@ async fn clear_overwritten_versions(agent: Agent) {
             }
         }
 
-        info!("got actors and their versions");
+        debug!("got actors and their versions");
 
         let cleared_versions: BTreeSet<i64> = {
             match pool.read().await {
@@ -1030,7 +1030,7 @@ async fn clear_overwritten_versions(agent: Agent) {
         let mut inserted = 0;
 
         for (actor_id, to_clear) in to_clear_by_actor {
-            info!(%actor_id, "clearing actor {} versions", to_clear.len());
+            info!(%actor_id, "Clearing {} versions", to_clear.len());
             let booked = {
                 bookie
                     .write(format!("to_clear_get_booked:{}", actor_id.as_simple()))
@@ -1063,7 +1063,7 @@ async fn clear_overwritten_versions(agent: Agent) {
         }
 
         info!(
-            "compaction done, cleared {} db bookkeeping table rows in {:?}",
+            "Compaction done, cleared {} DB bookkeeping table rows in {:?}",
             deleted - inserted,
             start.elapsed()
         );
@@ -1226,7 +1226,7 @@ fn find_cleared_db_versions(tx: &Transaction) -> rusqlite::Result<BTreeSet<i64>>
         .collect::<rusqlite::Result<_>>()?;
 
     info!(
-        "aggregated {} db versions to clear in {:?}",
+        "Aggregated {} DB versions to clear in {:?}",
         cleared_db_versions.len(),
         start.elapsed()
     );
@@ -1629,7 +1629,7 @@ async fn process_fully_buffered_changes(
 
         let tx = conn.transaction()?;
 
-        info!(%actor_id, version, "processing buffered changes to crsql_changes (actor: {actor_id}, version: {version}, last_seq: {last_seq})");
+        info!(%actor_id, version, "Processing buffered changes to crsql_changes (actor: {actor_id}, version: {version}, last_seq: {last_seq})");
 
         let max_db_version: Option<i64> = tx.prepare_cached("SELECT MAX(db_version) FROM __corro_buffered_changes WHERE site_id = ? AND version = ?")?.query_row(params![actor_id.as_bytes(), version], |row| row.get(0)).optional()?;
 
@@ -1649,9 +1649,9 @@ async fn process_fully_buffered_changes(
                             "#,
             )?
             .execute(params![max_db_version, actor_id.as_bytes(), version])?;
-            info!(%actor_id, version, "inserted {count} rows from buffered into crsql_changes in {:?}", start.elapsed());
+            info!(%actor_id, version, "Inserted {count} rows from buffered into crsql_changes in {:?}", start.elapsed());
         } else {
-            info!(%actor_id, version, "no buffered rows, skipped insertion into crsql_changes");
+            info!(%actor_id, version, "No buffered rows, skipped insertion into crsql_changes");
         }
 
         clear_buffered_meta(&tx, actor_id, version..=version)?;
@@ -1660,7 +1660,7 @@ async fn process_fully_buffered_changes(
             .prepare_cached("SELECT crsql_rows_impacted()")?
             .query_row((), |row| row.get(0))?;
 
-        info!(%actor_id, version, "rows impacted by buffered changes insertion: {rows_impacted}");
+        debug!(%actor_id, version, "rows impacted by buffered changes insertion: {rows_impacted}");
 
         let known_version = if rows_impacted > 0 {
             let db_version: i64 =
@@ -1687,7 +1687,7 @@ async fn process_fully_buffered_changes(
                 ":ts": ts
             })?;
 
-            info!(%actor_id, version, "inserted bookkeeping row after buffered insert");
+            debug!(%actor_id, version, "inserted bookkeeping row after buffered insert");
 
             Some(KnownDbVersion::Current {
                 db_version,
@@ -1697,7 +1697,7 @@ async fn process_fully_buffered_changes(
         } else {
             store_empty_changeset(&tx, actor_id, version..=version)?;
 
-            info!(%actor_id, version, "inserted CLEARED bookkeeping row after buffered insert");
+            debug!(%actor_id, version, "inserted CLEARED bookkeeping row after buffered insert");
             Some(KnownDbVersion::Cleared)
         };
 
@@ -1735,7 +1735,7 @@ pub async fn process_multiple_changes(
     agent: &Agent,
     changes: Vec<(ChangeV1, ChangeSource)>,
 ) -> Result<(), ChangeError> {
-    info!(self_actor_id = %agent.actor_id(), "processing multiple changes, len: {}", changes.iter().map(|(change, _)| cmp::max(change.len(), 1)).sum::<usize>());
+    debug!(self_actor_id = %agent.actor_id(), "processing multiple changes, len: {}", changes.iter().map(|(change, _)| cmp::max(change.len(), 1)).sum::<usize>());
 
     let bookie = agent.bookie();
 
@@ -1911,7 +1911,7 @@ pub async fn process_multiple_changes(
 
         tx.commit()?;
 
-        info!("committed {count} changes in {:?}", start.elapsed());
+        debug!("committed {count} changes in {:?}", start.elapsed());
 
         for (actor_id, knowns) in knowns {
             let booked = {
@@ -1934,7 +1934,7 @@ pub async fn process_multiple_changes(
                     let version = *versions.start();
                     if gaps_count == 0 {
                         // if we have no gaps, then we can schedule applying all these changes.
-                        info!(%actor_id, version, "we now have all versions, notifying for background jobber to insert buffered changes! seqs: {seqs:?}, expected full seqs: {full_seqs_range:?}");
+                        debug!(%actor_id, version, "we now have all versions, notifying for background jobber to insert buffered changes! seqs: {seqs:?}, expected full seqs: {full_seqs_range:?}");
                         let tx_apply = agent.tx_apply().clone();
                         tokio::spawn(async move {
                             if let Err(e) = tx_apply.send((actor_id, version)).await {
@@ -2082,7 +2082,7 @@ fn process_complete_version(
 
     let max_db_version = changes.iter().map(|c| c.db_version).max().unwrap_or(0);
 
-    info!(%actor_id, version, "complete change, applying right away! seqs: {seqs:?}, last_seq: {last_seq}, changes len: {len}, max db version: {max_db_version}");
+    debug!(%actor_id, version, "complete change, applying right away! seqs: {seqs:?}, last_seq: {last_seq}, changes len: {len}, max db version: {max_db_version}");
 
     debug_assert!(len <= (seqs.end() - seqs.start() + 1) as usize);
 
@@ -2258,8 +2258,6 @@ pub enum SyncClientError {
     Io(#[from] std::io::Error),
     #[error(transparent)]
     Pool(#[from] SqlitePoolError),
-    #[error("no good candidates found")]
-    NoGoodCandidate,
     #[error("could not decode message: {0}")]
     Decoded(#[from] SyncMessageDecodeError),
     #[error("could not encode message: {0}")]
@@ -2321,7 +2319,7 @@ async fn handle_sync(agent: &Agent, transport: &Transport) -> Result<(), SyncCli
         };
 
         if candidates.is_empty() {
-            return Err(SyncClientError::NoGoodCandidate);
+            return Ok(());
         }
 
         debug!("found {} candidates to synchronize with", candidates.len());
@@ -2351,7 +2349,7 @@ async fn handle_sync(agent: &Agent, transport: &Transport) -> Result<(), SyncCli
     };
 
     if chosen.is_empty() {
-        return Err(SyncClientError::NoGoodCandidate);
+        return Ok(());
     }
 
     let start = Instant::now();
@@ -2418,7 +2416,7 @@ async fn handle_changes(
         count = 0;
     }
 
-    info!("draining changes receiver...");
+    info!("Draining changes receiver...");
 
     // drain!
     while let Ok((change, src)) = rx_changes.try_recv() {
@@ -2487,7 +2485,7 @@ async fn write_empties_loop(
     }
 
     if !empties.is_empty() {
-        info!("inserting last unprocessed empties before shut down");
+        info!("Inserting last unprocessed empties before shut down");
         if let Err(e) = process_completed_empties(&agent, &mut empties).await {
             error!("could not process empties: {e}");
         }
@@ -2552,9 +2550,7 @@ async fn sync_loop(
                     }
                     tripwire::Outcome::Completed(res) => {
                         if let Err(e) = res {
-                            if !matches!(e, SyncClientError::NoGoodCandidate) {
-                                error!("could not sync: {e}");
-                            }
+                            error!("could not sync: {e}");
                             // keep syncing until we successfully sync
                             continue;
                         }
@@ -2565,13 +2561,13 @@ async fn sync_loop(
                     .reset(tokio::time::Instant::now() + sync_backoff.next().unwrap());
             }
             Branch::BackgroundApply { actor_id, version } => {
-                info!(%actor_id, version, "picked up background apply of buffered changes");
+                debug!(%actor_id, version, "picked up background apply of buffered changes");
                 match process_fully_buffered_changes(&agent, actor_id, version).await {
                     Ok(false) => {
                         warn!(%actor_id, version, "did not apply buffered changes");
                     }
                     Ok(true) => {
-                        info!(%actor_id, version, "succesfully applied buffered changes");
+                        debug!(%actor_id, version, "succesfully applied buffered changes");
                     }
                     Err(e) => {
                         error!(%actor_id, version, "could not apply fully buffered changes: {e}");
@@ -2587,7 +2583,7 @@ async fn process_completed_empties(
     agent: &Agent,
     empties: &mut BTreeMap<ActorId, RangeInclusiveSet<i64>>,
 ) -> eyre::Result<()> {
-    info!(
+    debug!(
         "processing empty versions (count: {})",
         empties.values().map(RangeInclusiveSet::len).sum::<usize>()
     );
@@ -2614,7 +2610,7 @@ async fn process_completed_empties(
         }
     }
 
-    info!(
+    debug!(
         "upserted {inserted} empty version ranges in {:?}",
         start.elapsed()
     );
@@ -3125,11 +3121,11 @@ pub mod tests {
 
                 let actual_count: i64 =
                     conn.query_row("SELECT count(*) FROM crsql_changes;", (), |row| row.get(0))?;
-                info!("actual count: {actual_count}");
+                debug!("actual count: {actual_count}");
 
                 let bookie = ta.agent.bookie();
 
-                info!(
+                debug!(
                     "last version: {:?}",
                     bookie
                         .write("test")
@@ -3144,7 +3140,6 @@ pub mod tests {
                 let needed = sync.need_len();
 
                 debug!("generated sync: {sync:?}");
-                info!("needed: {needed}");
 
                 v.push((counts.values().sum::<i64>(), needed));
             }

crates/corro-agent/src/api/peer.rs

@@ -451,7 +451,7 @@ fn handle_known_version(
                             Some(known) => match known {
                                 KnownVersion::Partial(PartialVersion { seqs, .. }) => {
                                     if seqs != &partial_seqs {
-                                        info!(%actor_id, version, "different partial sequences, updating! range_needed: {range_needed:?}");
+                                        warn!(%actor_id, version, "different partial sequences, updating! range_needed: {range_needed:?}");
                                         partial_seqs = seqs.clone();
                                         if let Some(new_start_seq) = last_sent_seq.take() {
                                             range_needed =
@@ -474,7 +474,7 @@ fn handle_known_version(
                         };
 
                         if let Some(known) = maybe_current_version {
-                            info!(%actor_id, version, "switched from partial to current version");
+                            warn!(%actor_id, version, "switched from partial to current version");
 
                             // drop write lock
                             drop(bw);
@@ -1053,7 +1053,7 @@ pub async fn parallel_sync(
 
                 trace!(%actor_id, "needs: {needs:?}");
 
-                info!(%actor_id, %addr, "needs len: {}", needs.values().map(|needs| needs.iter().map(|need| match need {
+                debug!(%actor_id, %addr, "needs len: {}", needs.values().map(|needs| needs.iter().map(|need| match need {
                     SyncNeedV1::Full {versions} => (versions.end() - versions.start()) as usize + 1,
                     SyncNeedV1::Partial {..} => 0,
                 }).sum::<usize>()).sum::<usize>());
@@ -1204,7 +1204,7 @@ pub async fn parallel_sync(
                     if let Err(e) = tx.finish().instrument(info_span!("quic_finish")).await {
                         warn!("could not finish stream while sending sync requests: {e}");
                     }
-                    info!(%server_actor_id, %addr, "done trying to sync w/ actor after {:?}", start.elapsed());
+                    debug!(%server_actor_id, %addr, "done trying to sync w/ actor after {:?}", start.elapsed());
                     continue;
                 }
 
@@ -1260,7 +1260,7 @@ pub async fn parallel_sync(
                 }
             }
 
-            info!(%actor_id, %count, "done reading sync messages");
+            debug!(%actor_id, %count, "done reading sync messages");
 
             Ok(count)
         }.instrument(info_span!("read_sync_requests_responses", %actor_id))
@@ -1289,7 +1289,7 @@ pub async fn serve_sync(
         opentelemetry::global::get_text_map_propagator(|propagator| propagator.extract(&trace_ctx));
     tracing::Span::current().set_parent(context);
 
-    info!(actor_id = %their_actor_id, self_actor_id = %agent.actor_id(), "received sync request");
+    debug!(actor_id = %their_actor_id, self_actor_id = %agent.actor_id(), "received sync request");
     let mut codec = LengthDelimitedCodec::new();
     let mut send_buf = BytesMut::new();
     let mut encode_buf = BytesMut::new();
@@ -1383,7 +1383,7 @@ pub async fn serve_sync(
                     stopped_res = write.stopped() => {
                         match stopped_res {
                             Ok(code) => {
-                                info!(actor_id = %their_actor_id, "send stream was stopped by peer, code: {code}");
+                                debug!(actor_id = %their_actor_id, "send stream was stopped by peer, code: {code}");
                             },
                             Err(e) => {
                                 warn!(actor_id = %their_actor_id, "error waiting for stop from stream: {e}");

crates/corro-agent/src/api/public/mod.rs

@@ -654,7 +654,7 @@ async fn execute_schema(agent: &Agent, statements: Vec<String>) -> eyre::Result<
             tx.execute("DELETE FROM __corro_schema WHERE tbl_name = ?", [tbl_name])?;
 
             let n = tx.execute("INSERT INTO __corro_schema SELECT tbl_name, type, name, sql, 'api' AS source FROM sqlite_schema WHERE tbl_name = ? AND type IN ('table', 'index') AND name IS NOT NULL AND sql IS NOT NULL", [tbl_name])?;
-            info!("updated {n} rows in __corro_schema for table {tbl_name}");
+            info!("Updated {n} rows in __corro_schema for table {tbl_name}");
         }
 
         tx.commit()?;

crates/corro-agent/src/api/public/pubsub.rs

@@ -137,7 +137,7 @@ pub async fn process_sub_channel(
             biased;
             Some(query_evt) = evt_rx.recv() => query_evt,
             _ = deadline_check => {
-                info!("all subscribers for {id} are gone and didn't come back within {MAX_UNSUB_TIME:?}");
+                info!("All subscribers for {id} are gone and didn't come back within {MAX_UNSUB_TIME:?}");
                 break;
             },
             _ = subs_check.tick() => {

crates/corro-client/src/lib.rs

@@ -185,7 +185,7 @@ impl CorrosionApiClient {
     pub async fn schema_from_paths<P: AsRef<Path>>(
         &self,
         schema_paths: &[P],
-    ) -> Result<ExecResponse, Error> {
+    ) -> Result<Option<ExecResponse>, Error> {
         let mut statements = vec![];
 
         for schema_path in schema_paths.iter() {
@@ -261,7 +261,11 @@ impl CorrosionApiClient {
             }
         }
 
-        self.schema(&statements).await
+        if statements.is_empty() {
+            return Ok(None);
+        }
+
+        Ok(Some(self.schema(&statements).await?))
     }
 }
 

crates/corro-types/src/agent.rs

@@ -303,7 +303,7 @@ impl SplitPool {
                 wait_conn_drop(tx).await
             }
 
-            info!("write loop done, draining...");
+            info!("Write loop done, draining...");
 
             // keep processing priority messages
             // NOTE: using `recv` would wait indefinitely, this loop only waits until all

crates/corro-types/src/pubsub.rs

@@ -689,7 +689,7 @@ impl Matcher {
 
                     match res {
                         Ok(deleted) => info!(
-                            "deleted {deleted} old changes row for subscription {}",
+                            "Deleted {deleted} old changes row for subscription {}",
                             self.id.as_simple()
                         ),
                         Err(e) => {

crates/corrosion/src/command/agent.rs

@@ -43,9 +43,12 @@ pub async fn run(config: Config, config_path: &Utf8PathBuf) -> eyre::Result<()>
             .schema_from_paths(config.db.schema_paths.as_slice())
             .await
         {
-            Ok(res) => {
+            Ok(Some(res)) => {
                 info!("Applied schema in {}s", res.time);
             }
+            Ok(None) => {
+                info!("No schema files to apply, skipping.");
+            }
             Err(e) => {
                 error!("could not apply schema: {e}");
             }

crates/corrosion/src/main.rs

@@ -202,7 +202,7 @@ async fn process_cli(cli: Cli) -> eyre::Result<()> {
                 )?;
             }
 
-            info!("successfully cleaned for restoration and backed up database to {path}");
+            info!("Successfully cleaned for restoration and backed up database to {path}");
         }
         Command::Restore {
             path,