Store and collector fixes from the review

CHANGELOG.md

@@ -8,7 +8,14 @@ its GitHub release, and Stackline shows it in the update dialog.
 ### Security
 - **SSH connections no longer open a port on this computer.** Stackline reached a server's Docker through a local TCP port that any program on the computer could also use, and a web page could try to. Whoever reaches that port controls the server's Docker. On macOS and Linux the connection now goes through a socket file in a folder only you can open, removed when Stackline stops. Windows still uses a local port for now
 
+### Changed
+- **A connection that reaches another server no longer deletes its logs by itself.** When the server behind a connection changes (its address was edited, an SSH alias points elsewhere, the swarm was set up again), Stackline stops collecting and asks: keep the old logs, or delete them. Before, they were deleted without asking
+- **At the size limit, a busy server no longer pushes out the logs of a quiet one.** Each connection gets an equal share of the space, and only connections above their share lose their oldest lines
+
 ### Fixed
+- **"Clear stored logs" during a read could leave part of that read behind**, and those lines then kept the cleared range from filling again
+- **A replica read for the first time held up all the others.** Reading hours of a busy replica can take minutes; other replicas and container events now go on being read meanwhile
+- **Deploys could go missing from the Overview** when a service crashed thousands of times after them
 - **A Docker that accepted a request and then never answered stopped log collection for good.** Requests now give up after 30 seconds without an answer and are tried again on the next poll
 
 ## 0.5.1

crates/stackline-core/src/collector.rs

@@ -53,6 +53,24 @@ pub struct Collector {
     store: Arc<Store>,
     states: HashMap<String, SourceState>,
     epoch: u64,
+    /// First reads under way, by source. They run apart from the regular
+    /// polls: reading hours of a busy replica can take minutes, and the other
+    /// replicas and the events must not wait for it.
+    first: HashMap<String, tokio::task::JoinHandle<(SourceState, usize)>>,
+}
+
+/// How many first reads of one connection run at once.
+const FIRST_READS: usize = 2;
+/// How long a poll waits for first reads to finish, so that a quick one
+/// (most of them) is stored by the poll that started it.
+const FIRST_WAIT: Duration = Duration::from_secs(2);
+
+impl Drop for Collector {
+    fn drop(&mut self) {
+        for (_, h) in self.first.drain() {
+            h.abort();
+        }
+    }
 }
 
 #[derive(Debug, Default, Clone)]
@@ -65,7 +83,7 @@ pub struct PollReport {
 impl Collector {
     pub fn new(conn: &str, store: Arc<Store>) -> Self {
         let epoch = store.epoch();
-        Self { conn: conn.to_string(), store, states: HashMap::new(), epoch }
+        Self { conn: conn.to_string(), store, states: HashMap::new(), epoch, first: HashMap::new() }
     }
 
     /// Reads new lines from every source. The first read of a new source goes
@@ -79,10 +97,18 @@ impl Collector {
         if refill {
             self.states.clear();
             self.epoch = epoch;
+            // What they read belongs to the lines just cleared.
+            for (_, h) in self.first.drain() {
+                h.abort();
+            }
         }
+        let started = Instant::now();
         let keys: std::collections::HashSet<String> = sources.iter().map(|s| s.key.clone()).collect();
         self.states.retain(|k, _| keys.contains(k));
         for src in sources {
+            if self.first.contains_key(&src.key) {
+                continue;
+            }
             let st = self.states.entry(src.key.clone()).or_insert_with(|| SourceState {
                 assembler: Assembler::new(&src.service),
                 cursor: None,
@@ -112,48 +138,90 @@ impl Collector {
             .filter(|(_, s)| s.src.running || s.last_read.is_none_or(|t| t.elapsed() > Duration::from_secs(60)))
             .map(|(k, _)| k.clone())
             .collect();
-        let mut taken: Vec<SourceState> = due.iter().filter_map(|k| self.states.remove(k)).collect();
+        let taken: Vec<SourceState> = due.iter().filter_map(|k| self.states.remove(k)).collect();
         let mut report = PollReport { sources: taken.len(), ..Default::default() };
+        let (fresh, known): (Vec<SourceState>, Vec<SourceState>) = taken.into_iter().partition(|st| st.cursor.is_none());
+
+        // First reads start in the background, a few at a time; the rest wait for a later poll.
+        for st in fresh {
+            if self.first.len() >= FIRST_READS {
+                self.states.insert(st.src.key.clone(), st);
+                continue;
+            }
+            let (conn, store, client) = (self.conn.clone(), self.store.clone(), client.clone());
+            let key = st.src.key.clone();
+            self.first.insert(key, tokio::spawn(read_one(client, store, conn, st, now, retention_us)));
+        }
 
         let conn = self.conn.clone();
         let store = self.store.clone();
-        let results: Vec<(SourceState, usize)> = futures_util::stream::iter(taken.drain(..))
-            .map(|mut st| {
-                let (conn, store, client) = (conn.clone(), store.clone(), client.clone());
-                async move {
-                    let n = match read_source(&client, &store, &conn, &mut st, now, retention_us).await {
-                        Ok(n) => {
-                            st.error = None;
-                            n
-                        }
-                        Err(e) => {
-                            st.error = Some(format!("{}: {e:#}", st.src.service));
-                            // Batches stored before the error stay stored: go on
-                            // after them, or the next poll stores them twice.
-                            // A half-built entry is dropped for the same reason.
-                            if st.stored > st.cursor {
-                                st.cursor = st.stored;
-                            }
-                            st.assembler = Assembler::new(&st.src.service);
-                            0
-                        }
-                    };
-                    st.last_read = Some(Instant::now());
-                    (st, n)
-                }
-            })
+        let results: Vec<(SourceState, usize)> = futures_util::stream::iter(known)
+            .map(|st| read_one(client.clone(), store.clone(), conn.clone(), st, now, retention_us))
             .buffer_unordered(6)
             .collect()
             .await;
         for (st, n) in results {
-            report.entries += n;
-            if let Some(e) = &st.error {
-                report.errors.push(e.clone());
+            self.done(st, n, &mut report);
+        }
+
+        // First reads that finished, waiting a moment for quick ones.
+        let deadline = started + FIRST_WAIT;
+        let keys: Vec<String> = self.first.keys().cloned().collect();
+        for key in keys {
+            let Some(h) = self.first.get_mut(&key) else { continue };
+            match tokio::time::timeout_at(deadline.into(), h).await {
+                Err(_) => continue,
+                Ok(r) => {
+                    self.first.remove(&key);
+                    // A read that panicked is started again from the stored cursor.
+                    if let Ok((st, n)) = r
+                        && st.epoch == self.epoch
+                    {
+                        self.done(st, n, &mut report);
+                    }
+                }
             }
-            self.states.insert(st.src.key.clone(), st);
         }
         report
     }
+
+    fn done(&mut self, st: SourceState, n: usize, report: &mut PollReport) {
+        report.entries += n;
+        if let Some(e) = &st.error {
+            report.errors.push(e.clone());
+        }
+        self.states.insert(st.src.key.clone(), st);
+    }
+}
+
+/// Reads one source and says how many entries it stored.
+async fn read_one(
+    client: DockerClient,
+    store: Arc<Store>,
+    conn: String,
+    mut st: SourceState,
+    now: i64,
+    retention_us: i64,
+) -> (SourceState, usize) {
+    let n = match read_source(&client, &store, &conn, &mut st, now, retention_us).await {
+        Ok(n) => {
+            st.error = None;
+            n
+        }
+        Err(e) => {
+            st.error = Some(format!("{}: {e:#}", st.src.service));
+            // Batches stored before the error stay stored: go on
+            // after them, or the next poll stores them twice.
+            // A half-built entry is dropped for the same reason.
+            if st.stored > st.cursor {
+                st.cursor = st.stored;
+            }
+            st.assembler = Assembler::new(&st.src.service);
+            0
+        }
+    };
+    st.last_read = Some(Instant::now());
+    (st, n)
 }
 
 async fn read_source(
@@ -306,19 +374,11 @@ async fn flush(store: &Arc<Store>, conn: &str, st: &mut SourceState, entries: &m
         (t, _) => t,
     };
     let (store, conn, src, epoch) = (store.clone(), conn.to_string(), st.src.clone(), st.epoch);
-    tokio::task::spawn_blocking(move || -> Result<()> {
-        // The store was cleared during this read: the next poll starts over.
-        if store.epoch() != epoch {
-            return Ok(());
-        }
-        store.insert(&conn, &src, &batch)?;
-        if let Some(ts) = last_ts {
-            store.set_cursor(&conn, &src.key, ts)?;
-        }
-        Ok(())
-    })
-    .await??;
-    st.stored = last_ts.or(st.stored);
+    // A store cleared during this read keeps nothing of it: the next poll starts over.
+    let stored = tokio::task::spawn_blocking(move || store.store_batch(&conn, &src, &batch, last_ts, epoch)).await??;
+    if stored {
+        st.stored = last_ts.or(st.stored);
+    }
     Ok(n)
 }
 
@@ -485,6 +545,66 @@ mod tests {
         assert_eq!(stored(&store), 5);
     }
 
+    #[tokio::test]
+    async fn a_slow_first_read_does_not_hold_up_the_other_replicas() {
+        // Answers every log request at once with one new line, except those of
+        // the replica `slow`, which take 4 seconds: a first read of hours of logs.
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap().to_string();
+        let line = Arc::new(Mutex::new(0i64));
+        let l = line.clone();
+        tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener.accept().await else { return };
+                let l = l.clone();
+                tokio::spawn(async move {
+                    let mut buf = Vec::new();
+                    let mut chunk = [0u8; 1024];
+                    while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
+                        match sock.read(&mut chunk).await {
+                            Ok(0) | Err(_) => return,
+                            Ok(n) => buf.extend_from_slice(&chunk[..n]),
+                        }
+                    }
+                    let head = String::from_utf8_lossy(&buf).to_string();
+                    if head.contains("/slow/") {
+                        tokio::time::sleep(Duration::from_secs(4)).await;
+                    }
+                    let ts = *l.lock().unwrap();
+                    let body = frames(&[(ts, "INFO hello".to_string())], "/");
+                    let res = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len());
+                    let _ = sock.write_all(res.as_bytes()).await;
+                    let _ = sock.write_all(&body).await;
+                });
+            }
+        });
+        let client = DockerClient::new(Endpoint::Tcp(addr));
+        let store = Arc::new(Store::open_in_memory().unwrap());
+        let now = chrono::Utc::now().timestamp_micros();
+        let fast = LogSource { key: "fast".into(), ..source() };
+        let slow = LogSource { key: "slow".into(), service: "shop_worker".into(), ..source() };
+        let mut c = Collector::new("conn", store.clone());
+
+        *line.lock().unwrap() = now - 10_000_000;
+        let started = Instant::now();
+        let r = c.poll(&client, vec![fast.clone(), slow.clone()], RETENTION).await;
+        assert!(started.elapsed() < Duration::from_secs(3), "the poll waited {:?} for the slow first read", started.elapsed());
+        assert_eq!(r.entries, 1, "{r:?}");
+
+        // The next polls read the fast replica again while the slow one is still busy.
+        *line.lock().unwrap() = now - 5_000_000;
+        let started = Instant::now();
+        let r = c.poll(&client, vec![fast.clone(), slow.clone()], RETENTION).await;
+        assert!(started.elapsed() < Duration::from_secs(3));
+        assert_eq!(r.entries, 1, "{r:?}");
+
+        // Once done, the slow replica's lines are stored and it is polled like any other.
+        tokio::time::sleep(Duration::from_secs(3)).await;
+        c.poll(&client, vec![fast, slow], RETENTION).await;
+        assert!(store.cursor("conn", "slow").unwrap().is_some());
+        assert_eq!(stored(&store), 3);
+    }
+
     #[tokio::test]
     async fn a_new_replica_is_read_6_hours_back_and_a_clear_refills_the_retention() {
         let docker = FakeDocker::start().await;

crates/stackline-core/src/events.rs

@@ -132,16 +132,9 @@ impl EventPoller {
         events.extend(snap.task_failures.iter().filter(|t| t.node != names.local || t.container.is_empty()).map(from_task));
 
         let (store, conn) = (self.store.clone(), self.conn.clone());
-        tokio::task::spawn_blocking(move || {
-            // A clear while reading: what was read belongs to the lines just deleted.
-            if store.epoch() != epoch {
-                return Ok(0);
-            }
-            let n = store.insert_events(&conn, &events)?;
-            store.set_cursor(&conn, CURSOR, now)?;
-            Ok(n)
-        })
-        .await?
+        // A clear while reading: what was read belongs to the lines just deleted.
+        let n = tokio::task::spawn_blocking(move || store.store_events(&conn, &events, Some(now), epoch)).await??;
+        Ok(n.unwrap_or(0))
     }
 }
 

crates/stackline-core/src/runtime.rs

@@ -17,7 +17,7 @@ use crate::docker::{DockerClient, Endpoint};
 use crate::events::EventPoller;
 use crate::model::Snapshot;
 use crate::snapshot::{self, Options};
-use crate::store::Store;
+use crate::store::{HostCheck, Store};
 
 #[derive(Clone, Debug, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "lowercase")]
@@ -37,6 +37,10 @@ pub struct ConnState {
     pub updated_at: Option<i64>,
     pub logs_synced_at: Option<i64>,
     pub log_errors: Vec<String>,
+    /// The connection now reaches another host or swarm than its stored
+    /// lines come from. Nothing is collected until the user decides whether
+    /// the old lines stay ([`crate::store::Store::accept_host`]).
+    pub other_host: bool,
 }
 
 /// One opened connection.
@@ -65,6 +69,7 @@ impl Runtime {
                 updated_at: None,
                 logs_synced_at: None,
                 log_errors: vec![],
+                other_host: false,
             }),
             client: tokio::sync::Mutex::new(None),
             tasks: Mutex::new(vec![]),
@@ -81,6 +86,16 @@ impl Runtime {
         self.state.read().unwrap().clone()
     }
 
+    /// The host this connection reaches now, as the store remembers hosts.
+    pub fn identity(&self) -> Option<String> {
+        self.snapshot()?.identity()
+    }
+
+    /// The user answered the question about another host: collecting goes on.
+    pub fn host_accepted(&self) {
+        self.state.write().unwrap().other_host = false;
+    }
+
     /// Reads the full configuration of a service by its full name.
     pub async fn details(&self, name: &str) -> Result<Option<crate::details::ServiceDetails>> {
         let Some(snap) = self.snapshot() else { return Ok(None) };
@@ -328,14 +343,18 @@ impl Manager {
                     // Checked when the identity is first seen or changes, not on every poll.
                     if let Some(identity) = snap.identity().filter(|i| bound.as_ref() != Some(i)) {
                         let (store, conn, id2) = (m.store.clone(), r.cfg.id.clone(), identity.clone());
-                        // Clearing a large store takes a while; keep it off the async workers.
-                        match tokio::task::spawn_blocking(move || store.bind_host(&conn, &id2)).await {
-                            Ok(Ok(cleared)) => {
-                                if cleared {
-                                    tracing::info!("{} now reaches another host; its stored logs were cleared", r.cfg.name);
-                                    collector = Collector::new(&r.cfg.id, m.store.clone());
-                                    events = EventPoller::new(&r.cfg.id, m.store.clone());
+                        match tokio::task::spawn_blocking(move || store.check_host(&conn, &id2)).await {
+                            Ok(Ok(HostCheck::Changed)) => {
+                                // Nothing is read until the user says what happens to the old
+                                // lines; asked again on every poll until then.
+                                if !std::mem::replace(&mut r.state.write().unwrap().other_host, true) {
+                                    tracing::warn!("{} now reaches another host; collecting waits for your answer", r.cfg.name);
                                 }
+                                tokio::time::sleep(Duration::from_secs(settings.log_poll_secs.max(1) as u64)).await;
+                                continue;
+                            }
+                            Ok(Ok(_)) => {
+                                r.state.write().unwrap().other_host = false;
                                 bound = Some(identity);
                             }
                             Ok(Err(e)) => tracing::warn!("could not check the host of {}: {e:#}", r.cfg.name),