fix(reload): start, stop and rebuild sources on a config reload

crates/core/src/lib.rs

@@ -357,6 +357,12 @@ pub trait RowSource: Send + Sync {
 
     /// Health check: credentials resolve and the provider answers.
     fn probe(&self) -> BoxFuture<'_, Result<()>>;
+
+    /// Stop what the source runs between ticks and wait for it to finish.
+    /// Its driver calls this when it stops, possibly more than once.
+    fn stop(&self) -> BoxFuture<'_, ()> {
+        Box::pin(std::future::ready(()))
+    }
 }
 
 #[cfg(test)]

crates/fetcher/src/driver.rs

@@ -410,11 +410,14 @@ impl Driver {
     /// Stop what the shape runs in the background and wait for it to finish.
     ///
     /// A file tailer writes its final checkpoint, including the last
-    /// acknowledged tick, which only the driver's checkpoint store holds.
-    /// Every other shape runs nothing between ticks. Safe to call again.
+    /// acknowledged tick, which only the driver's checkpoint store holds. A
+    /// custom shape runs its own [`RowSource::stop`], and a REST or database
+    /// shape runs nothing between ticks. Safe to call again.
     pub async fn stop(&self) {
-        let Shape::File(shape) = &self.shape else {
-            return;
+        let shape = match &self.shape {
+            Shape::File(shape) => shape,
+            Shape::Custom(source) => return source.stop().await,
+            Shape::Rest(_) | Shape::Db(_) => return,
         };
         let mut committed = HashMap::new();
         for unit in shape.units() {