andrey/zaloopipe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: persist run state to disk with recovery on startup

Browse Source
This commit is contained in:
andrey
2026-07-16 10:23:03 +03:00
parent db11d7a4d9
commit e37545ffff
4 changed files with 310 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pipeline-runner/src/runner.cpp
+155 -4
View File
@@ -2,11 +2,90 @@
#include "pipeline.h"
#include <pidir.h>
#include <pifile.h>
#include <pijson.h>
#include <piprocess.h>
#include <pisystemtime.h>
#include <pithread.h>
#include <pitime.h>
static PIString getStateFilePath(const PIString & logs_dir, const PIString & run_id) {
return logs_dir + run_id + ".state.json";
}
PIJSON PipelineRunner::stepResultToJSON(const StepResult & sr) {
PIJSON j = PIJSON::newObject();
j["step_index"] = sr.step_index;
j["title"] = sr.title;
j["status"] = runStatusToString(sr.status);
j["returncode"] = sr.returncode;
j["output"] = sr.output;
j["error"] = sr.error;
return j;
}
StepResult PipelineRunner::stepResultFromJSON(const PIJSON & j) {
StepResult sr;
sr.step_index = j["step_index"].toInt();
sr.title = j["title"].toString();
PIString statusStr = j["status"].toString();
if (statusStr == "pending")
sr.status = RunStatus::Pending;
else if (statusStr == "running")
sr.status = RunStatus::Running;
else if (statusStr == "completed")
sr.status = RunStatus::Completed;
else
sr.status = RunStatus::Error;
sr.returncode = j["returncode"].toInt();
sr.output = j["output"].toString();
sr.error = j["error"].toString();
return sr;
}
PIJSON PipelineRunner::runStateToJSON(const RunState & state) {
PIJSON j = PIJSON::newObject();
j["run_id"] = state.run_id;
j["pipeline_id"] = state.pipeline_id;
j["status"] = runStatusToString(state.status);
j["current_step"] = state.current_step;
PIJSON stepsArr = PIJSON::newArray();
for (int i = 0; i < state.steps.size(); ++i) {
stepsArr << stepResultToJSON(state.steps[i]);
}
j["steps"] = stepsArr;
return j;
}
RunState PipelineRunner::runStateFromJSON(const PIJSON & j) {
RunState state;
state.run_id = j["run_id"].toString();
state.pipeline_id = j["pipeline_id"].toString();
PIString statusStr = j["status"].toString();
if (statusStr == "pending")
state.status = RunStatus::Pending;
else if (statusStr == "running")
state.status = RunStatus::Running;
else if (statusStr == "completed")
state.status = RunStatus::Completed;
else
state.status = RunStatus::Error;
state.current_step = j["current_step"].toInt();
PIJSON stepsArrJSON = j["steps"];
const auto & stepsArr = stepsArrJSON.array();
for (int i = 0; i < stepsArr.size(); ++i) {
if (stepsArr[i].isObject()) {
state.steps << stepResultFromJSON(stepsArr[i]);
}
}
return state;
}
PipelineRunner::PipelineRunner(const PIString & pipelines_dir, const PIString & logs_dir)
: pipelines_dir_(pipelines_dir)
, logs_dir_(logs_dir) {}
@@ -47,6 +126,7 @@ PIString PipelineRunner::startRun(const PIString & pipeline_id) {
active_runs_.insert(run_id, true);
}
saveRunState(run_id);
appendLog(run_id, "Starting pipeline: " + pipeline.name + " (run_id: " + run_id + ")");
// Execute in a separate thread
@@ -59,11 +139,19 @@ PIString PipelineRunner::startRun(const PIString & pipeline_id) {
}
RunState PipelineRunner::getRunState(const PIString & run_id) {
PIMutexLocker ml(mutex_);
if (runs_.contains(run_id)) {
return runs_.value(run_id);
{
PIMutexLocker ml(mutex_);
if (runs_.contains(run_id)) {
return runs_.value(run_id);
}
}
return RunState();
RunState state = loadRunState(run_id);
if (!state.run_id.isEmpty()) {
PIMutexLocker ml(mutex_);
runs_.insert(run_id, state);
}
return state;
}
bool PipelineRunner::isRunActive(const PIString & run_id) {
@@ -100,6 +188,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
active_runs_.remove(run_id);
}
saveRunState(run_id);
appendLog(run_id, "Pipeline finished");
#else
for (int i = 0; i < total; ++i) {
@@ -116,6 +206,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
}
}
saveRunState(run_id);
// Build command: opencode run <text> --title <title>
PIStringList args;
args << "run" << prompt.text << "--title" << prompt.title;
@@ -141,6 +233,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
}
}
saveRunState(run_id);
if (rc != 0) {
appendLog(run_id, "Pipeline failed at step " + PIString::fromNumber(i + 1));
{
@@ -149,6 +243,7 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
runs_[run_id].status = RunStatus::Error;
}
}
saveRunState(run_id);
break;
}
}
@@ -162,6 +257,8 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
active_runs_.remove(run_id);
}
saveRunState(run_id);
appendLog(run_id, "Pipeline finished");
#endif
}
@@ -169,3 +266,57 @@ void PipelineRunner::executePipeline(const PIString & run_id, const Pipeline & p
void PipelineRunner::appendLog(const PIString & run_id, const PIString & line) {
::appendLog(logs_dir_, run_id, line);
}
void PipelineRunner::saveRunState(const PIString & run_id) {
PIMutexLocker ml(mutex_);
if (!runs_.contains(run_id)) return;
RunState state = runs_.value(run_id);
PIJSON j = runStateToJSON(state);
PIString path = getStateFilePath(logs_dir_, run_id);
PIString tmpPath = path + ".tmp";
if (PIFile::writeAll(tmpPath, j.toJSON(PIJSON::Tree).toUTF8())) {
PIFile::remove(path);
PIFile::rename(tmpPath, path);
}
}
RunState PipelineRunner::loadRunState(const PIString & run_id) {
PIString path = getStateFilePath(logs_dir_, run_id);
if (!PIFile::isExists(path)) return RunState();
PIByteArray data = PIFile::readAll(path);
PIJSON j = PIJSON::fromJSON(PIString::fromUTF8(data));
if (j.isObject()) {
return runStateFromJSON(j);
}
return RunState();
}
void PipelineRunner::recoverRuns() {
PIDir dir(logs_dir_);
if (!dir.isExists()) return;
const auto entries = dir.entries();
for (const auto & entry: entries) {
if (!entry.isFile() || entry.extension() != "json") continue;
PIString fileName = entry.name();
if (!fileName.endsWith(".state.json")) continue;
PIString run_id = fileName.left(fileName.length() - PIString(".state.json").length());
RunState state = loadRunState(run_id);
if (state.status == RunStatus::Running) {
PIMutexLocker ml(mutex_);
runs_.insert(run_id, state);
active_runs_.insert(run_id, true);
}
}
}
void PipelineRunner::removeStateFile(const PIString & run_id) {
PIString path = getStateFilePath(logs_dir_, run_id);
PIFile::remove(path);
}
pipeline-runner/src/runner.h
+10
View File
@@ -60,7 +60,17 @@ public:
// Check if run is still active
bool isRunActive(const PIString & run_id);
// Persist run state to disk
void saveRunState(const PIString & run_id);
RunState loadRunState(const PIString & run_id);
void recoverRuns();
void removeStateFile(const PIString & run_id);
private:
PIJSON runStateToJSON(const RunState & state);
PIJSON stepResultToJSON(const StepResult & sr);
RunState runStateFromJSON(const PIJSON & j);
StepResult stepResultFromJSON(const PIJSON & j);
void executePipeline(const PIString & run_id, const Pipeline & pipeline);
void appendLog(const PIString & run_id, const PIString & line);
pipeline-runner/src/server.cpp
+3
View File
@@ -28,6 +28,9 @@ bool Server::start() {
PIDir::make(pipelines_dir_);
PIDir::make(logs_dir_);
// Recover running state from disk
runner_.recoverRuns();
// Register routes
httpserver_->registerPath("/api/pipelines", PIHTTP::Method::Get, this, &Server::listPipelines);
httpserver_->registerPath("/api/pipelines", PIHTTP::Method::Post, this, &Server::createPipeline);
pipeline-runner/tests/test_runner.cpp
+142
View File
@@ -151,6 +151,148 @@ TEST(RunnerTest, RunStatusToString) {
EXPECT_EQ(runStatusToString(RunStatus::Error), "error");
}
TEST(PersistenceTest, SaveLoadRoundTrip) {
setupDirs();
PipelineRunner runner(testPipelinesDir, testLogsDir);
// Test loadRunState on non-existent file
RunState empty = runner.loadRunState("nonexistent");
EXPECT_TRUE(empty.run_id.isEmpty());
// Create state file with one step
PIString runId = "test-run-123";
PIString statePath = testLogsDir + runId + ".state.json";
PIJSON j = PIJSON::newObject();
j["run_id"] = runId;
j["pipeline_id"] = "test-pipeline";
j["status"] = "running";
j["current_step"] = 0;
PIJSON stepsArr = PIJSON::newArray();
PIJSON sj = PIJSON::newObject();
sj["step_index"] = 0;
sj["title"] = "Step 1";
sj["status"] = "completed";
sj["returncode"] = 0;
sj["output"] = "output data";
sj["error"] = "";
stepsArr << sj;
j["steps"] = stepsArr;
PIFile::writeAll(statePath, j.toJSON(PIJSON::Tree).toUTF8());
// Load via runner
RunState loaded = runner.loadRunState(runId);
EXPECT_EQ(loaded.run_id, runId);
EXPECT_EQ(loaded.pipeline_id, "test-pipeline");
EXPECT_EQ(loaded.status, RunStatus::Running);
EXPECT_EQ(loaded.steps.size(), 1);
EXPECT_EQ(loaded.steps[0].title, "Step 1");
EXPECT_EQ(loaded.steps[0].status, RunStatus::Completed);
cleanupDirs();
}
TEST(PersistenceTest, RecoverRuns) {
setupDirs();
// Create state file manually to simulate a running state from a previous session
PIString runId = "recovered-run-456";
PIString statePath = testLogsDir + runId + ".state.json";
PIJSON j = PIJSON::newObject();
j["run_id"] = runId;
j["pipeline_id"] = "some-pipeline";
j["status"] = "running";
j["current_step"] = 1;
PIJSON stepsArr = PIJSON::newArray();
PIJSON sj = PIJSON::newObject();
sj["step_index"] = 0;
sj["title"] = "Step 1";
sj["status"] = "completed";
sj["returncode"] = 0;
sj["output"] = "";
sj["error"] = "";
stepsArr << sj;
PIJSON sj2 = PIJSON::newObject();
sj2["step_index"] = 1;
sj2["title"] = "Step 2";
sj2["status"] = "pending";
sj2["returncode"] = 0;
sj2["output"] = "";
sj2["error"] = "";
stepsArr << sj2;
j["steps"] = stepsArr;
PIFile::writeAll(statePath, j.toJSON(PIJSON::Tree).toUTF8());
// Also create a completed state file that should NOT be recovered
PIString completedRunId = "completed-run-789";
PIString completedStatePath = testLogsDir + completedRunId + ".state.json";
PIJSON j2 = PIJSON::newObject();
j2["run_id"] = completedRunId;
j2["pipeline_id"] = "some-pipeline";
j2["status"] = "completed";
j2["current_step"] = 0;
j2["steps"] = PIJSON::newArray();
PIFile::writeAll(completedStatePath, j2.toJSON(PIJSON::Tree).toUTF8());
// New runner should recover only the running state
PipelineRunner runner(testPipelinesDir, testLogsDir);
runner.recoverRuns();
RunState recovered = runner.getRunState(runId);
EXPECT_EQ(recovered.run_id, runId);
EXPECT_EQ(recovered.status, RunStatus::Running);
EXPECT_EQ(recovered.steps.size(), 2);
// Completed run should NOT be in memory after recovery
EXPECT_FALSE(runner.isRunActive(completedRunId));
cleanupDirs();
}
TEST(PersistenceTest, CompletedRunStatePersisted) {
setupDirs();
PipelineRunner runner(testPipelinesDir, testLogsDir);
Pipeline pl;
pl.id = "persist-test";
pl.name = "Persist Test";
pl.working_dir = "/tmp";
pl.created_at = nowISO();
pl.updated_at = pl.created_at;
Prompt p;
p.id = "p1";
p.text = "test";
p.title = "Step 1";
p.order = 0;
pl.prompts << p;
savePipeline(testPipelinesDir, pl);
PIString runId = runner.startRun("persist-test");
EXPECT_FALSE(runId.isEmpty());
waitForRun(runner, runId);
// Verify state file exists and contains final status
PIString statePath = testLogsDir + runId + ".state.json";
EXPECT_TRUE(PIFile::isExists(statePath));
RunState loaded = runner.loadRunState(runId);
EXPECT_EQ(loaded.run_id, runId);
cleanupDirs();
}
#ifdef FAKE
static Pipeline makeFakePipeline(const PIString & id, const PIString & name, int numSteps) {