Introduced command "Replay"

Author: spirali

crates/hyperqueue/src/client/commands/journal/mod.rs

@@ -25,8 +25,13 @@ enum JournalCommand {
     /// Events will be exported to `stdout`, you can redirect it e.g. to a file.
     Export(ExportOpts),
 
-    /// Stream events from a running server.
-    Stream(StreamOpts),
+    /// Stream events from a running server, it first replays old events
+    /// then it waits for new live events.
+    Stream,
+
+    /// Stream events from a running server, it replays old events
+    /// after that it terminates the connection.
+    Replay,
 
     /// Connect to a server and remove completed tasks and non-active workers from journal
     Prune,
@@ -43,29 +48,22 @@ struct ExportOpts {
     journal: PathBuf,
 }
 
-#[derive(Parser)]
-struct StreamOpts {
-    /// If enabled, server terminates the connection when all currents events
-    /// are sent.
-    #[arg(long)]
-    history_only: bool,
-}
-
 pub async fn command_journal(gsettings: &GlobalSettings, opts: JournalOpts) -> anyhow::Result<()> {
     match opts.command {
         JournalCommand::Export(opts) => export_json(opts),
-        JournalCommand::Stream(opts) => stream_json(gsettings, opts).await,
+        JournalCommand::Replay => stream_json(gsettings, false).await,
+        JournalCommand::Stream => stream_json(gsettings, true).await,
         JournalCommand::Prune => prune_journal(gsettings).await,
         JournalCommand::Flush => flush_journal(gsettings).await,
     }
 }
 
-async fn stream_json(gsettings: &GlobalSettings, opts: StreamOpts) -> anyhow::Result<()> {
+async fn stream_json(gsettings: &GlobalSettings, live_events: bool) -> anyhow::Result<()> {
     let mut connection = get_client_session(gsettings.server_directory()).await?;
     connection
         .connection()
         .send(FromClientMessage::StreamEvents(StreamEvents {
-            history_only: opts.history_only,
+            live_events,
         }))
         .await?;
     let stdout = std::io::stdout();

crates/hyperqueue/src/dashboard/data/fetch.rs

@@ -11,7 +11,7 @@ pub async fn create_data_fetch_process(
     session
         .connection()
         .send(FromClientMessage::StreamEvents(StreamEvents {
-            history_only: false,
+            live_events: true,
         }))
         .await?;
 

crates/hyperqueue/src/server/client/mod.rs

@@ -175,7 +175,7 @@ pub async fn client_rpc_loop<
                         So while historic events are loaded from the file and streamed, live events are already
                         collected and sent immediately the historic events are sent */
                         let (tx1, rx1) = mpsc::unbounded_channel::<Event>();
-                        let live = if !msg.history_only {
+                        let live = if msg.live_events {
                             let (tx2, rx2) = mpsc::unbounded_channel::<Event>();
                             let listener_id = senders.events.register_listener(tx2);
                             Some((rx2, listener_id))

crates/hyperqueue/src/transfer/messages.rs

@@ -88,7 +88,7 @@ pub struct TaskKindProgram {
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct StreamEvents {
-    pub history_only: bool,
+    pub live_events: bool,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]

tests/test_events.py

@@ -64,10 +64,10 @@ def test_worker_stream_events2(hq_env: HqEnv, tmp_path):
         assert events[2]["event"]["desc"]["name"] == "uname"
 
 
-def test_worker_stream_history_only(hq_env: HqEnv, tmp_path):
+def test_worker_journal_replay(hq_env: HqEnv, tmp_path):
     journal = tmp_path.joinpath("test.journal")
     hq_env.start_server(args=["--journal", journal])
-    r = hq_env.command(["journal", "stream", "--history-only"])
+    r = hq_env.command(["journal", "replay"])
     msg = json.loads(r)
     assert msg["event"]["type"] == "server-start"
 
@@ -162,21 +162,21 @@ def body():
         with hq_env.mock.mock_program_with_code(
             "rocm-smi",
             """
-    import json
-    data = {
-        "card0": {
-            "GPU use (%)": "1.5",
-            "GPU memory use (%)": "12.5",
-            "PCI Bus": "FOOBAR1"
-        },
-        "card1": {
-            "GPU use (%)": "12.5",
-            "GPU memory use (%)": "64.0",
-            "PCI Bus": "FOOBAR2"
-        }
+import json
+data = {
+    "card0": {
+        "GPU use (%)": "1.5",
+        "GPU memory use (%)": "12.5",
+        "PCI Bus": "FOOBAR1"
+    },
+    "card1": {
+        "GPU use (%)": "12.5",
+        "GPU memory use (%)": "64.0",
+        "PCI Bus": "FOOBAR2"
     }
-    print(json.dumps(data))
-            """,
+}
+print(json.dumps(data))
+""",
         ):
             hq_env.start_worker(args=["--overview-interval", "10ms", "--resource", "gpus/amd=[0]"])
             wait_for_worker_state(hq_env, 1, "RUNNING")