From ef435148f86e5024c4534cedc77dcf94b93e9146 Mon Sep 17 00:00:00 2001 From: fy59 Date: Thu, 17 Sep 2026 13:06:54 +0200 Subject: [PATCH] feat(sync): add durable web orchestration API --- tests/sync_orchestrator_peer_fixture.py | 16 ++ tests/test_sync_orchestrator.py | 31 +++ tools/sync_orchestrator.py | 179 +++++++++++++++++ tools/validate_sync_isolated.py | 7 +- tui/include/trainlog/web_server.h | 3 +- tui/meson.build | 9 + tui/src/main.c | 7 +- tui/src/web_server.c | 255 +++++++++++++++++++++++- tui/tests/test_web_server.c | 30 ++- 9 files changed, 526 insertions(+), 11 deletions(-) create mode 100644 tests/sync_orchestrator_peer_fixture.py create mode 100644 tests/test_sync_orchestrator.py create mode 100644 tools/sync_orchestrator.py diff --git a/tests/sync_orchestrator_peer_fixture.py b/tests/sync_orchestrator_peer_fixture.py new file mode 100644 index 0000000..5c2a48e --- /dev/null +++ b/tests/sync_orchestrator_peer_fixture.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python3 +"""Bounded isolated peer transport fixture for orchestrator/Web tests.""" +import json, os, uuid + +run_id=os.environ["TRAINLOG_SYNC_RUN_ID"] +def identity(prefix): return prefix+str(uuid.uuid4()) +print(json.dumps({ + "format":"trainlog-sync-worker-report","version":1,"run_id":run_id, + "producer_peer_id":identity("peer_"),"consumer_peer_id":identity("peer_"), + "inbound_generation_id":identity("gen_"),"outbound_generation_id":identity("gen_"), + "manifest_sha256":"a"*64,"result":"completed","sessions_reconciled":1, + "domains":{"history-v4":True,"execution-draft-v1":True,"causal-delete-v1":True, + "equipment":True,"aliases":True,"profiles":True,"body-zones":True,"feedback":True}, + "drafts":[{"draft_id":identity("draft_"),"state":"pending","session_type":"strength","occurrence_count":2}], + "ai_midpoint":{"result":"not_configured"},"ai_post_sync":{"result":"not_configured"} +},sort_keys=True,separators=(",",":"))) diff --git a/tests/test_sync_orchestrator.py b/tests/test_sync_orchestrator.py new file mode 100644 index 0000000..50ec59a --- /dev/null +++ b/tests/test_sync_orchestrator.py @@ -0,0 +1,31 @@ +import json, os, subprocess, tempfile, unittest, uuid +from pathlib import Path + +ROOT=Path(__file__).resolve().parents[1] +ORCHESTRATOR=ROOT/"tools/sync_orchestrator.py" +FIXTURE=ROOT/"tests/sync_orchestrator_peer_fixture.py" + +class OrchestratorTest(unittest.TestCase): + def test_correlated_worker_report_is_durable_and_reopenable(self): + for trigger in ("web","android"): + with self.subTest(trigger=trigger), tempfile.TemporaryDirectory() as directory: + root=Path(directory);db=root/"trainlog.db";db.touch();state=Path(str(db)+".sync-run.json") + run="sy_"+str(uuid.uuid4());request="sy_"+str(uuid.uuid4()) + state.write_text(json.dumps({"run_id":run,"request_id":request,"trigger":trigger,"phase":"requested","progress_revision":1,"result":"running"})) + config=root/"config.json";config.write_text(json.dumps({"format":"trainlog-sync-orchestrator-config","version":1,"enabled":True,"mode":"directory-test-double","capabilities":["generation-manifest-v1","generation-ack-v1","causal-delete-v1","mobile-history-v4","execution-draft-v1"],"peer_command":["python3",str(FIXTURE)],"timeout_seconds":30})) + result=subprocess.run(["python3",str(ORCHESTRATOR),"--database",str(db),"--state",str(state),"--config",str(config),"--run-id",run,"--request-id",request],capture_output=True,text=True) + self.assertEqual(result.returncode,0,result.stderr) + final=json.loads(state.read_text());self.assertEqual(final["phase"],"completed") + self.assertEqual(final["result"],"completed");self.assertEqual(final["sessions_reconciled"],1) + self.assertTrue(final["domains"]["causal-delete-v1"]);self.assertEqual(final["drafts"][0]["state"],"pending") + + def test_missing_capability_is_explicitly_degraded(self): + with tempfile.TemporaryDirectory() as directory: + root=Path(directory);db=root/"trainlog.db";db.touch();state=Path(str(db)+".sync-run.json") + run="sy_"+str(uuid.uuid4());request="sy_"+str(uuid.uuid4()) + state.write_text(json.dumps({"run_id":run,"request_id":request,"trigger":"web","phase":"requested","progress_revision":1,"result":"running"})) + config=root/"config.json";config.write_text(json.dumps({"format":"trainlog-sync-orchestrator-config","version":1,"enabled":True,"mode":"directory-test-double","capabilities":[],"peer_command":["python3",str(FIXTURE)],"timeout_seconds":30})) + result=subprocess.run(["python3",str(ORCHESTRATOR),"--database",str(db),"--state",str(state),"--config",str(config),"--run-id",run,"--request-id",request]) + self.assertEqual(result.returncode,3);self.assertEqual(json.loads(state.read_text())["phase"],"explicitly_degraded") + +if __name__=="__main__":unittest.main() diff --git a/tools/sync_orchestrator.py b/tools/sync_orchestrator.py new file mode 100644 index 0000000..7931501 --- /dev/null +++ b/tools/sync_orchestrator.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 +"""Durable full-generation synchronization run coordinator. + +The HTTP adapter starts this executable with fixed arguments. Peer transport is +selected only by a trusted local configuration file; request JSON can never +choose commands, paths, databases, capabilities, or causal policy. +""" +from __future__ import annotations + +import argparse +import fcntl +import json +import os +import subprocess +import signal +import sys +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +MAX_CONFIG = 16 * 1024 +MAX_REPORT = 64 * 1024 +MAX_DIAGNOSTIC = 1024 +PHASES = {"requested", "waiting_android_publication", "running", + "local_import_committed", "published", "waiting_acknowledgement", + "peer_consumed", "completed", "failed", "interrupted", "explicitly_degraded"} + + +def canonical(value: object) -> bytes: + return (json.dumps(value, ensure_ascii=False, sort_keys=True, + separators=(",", ":")) + "\n").encode() + + +def atomic_write(path: Path, value: dict) -> None: + raw = canonical(value) + if len(raw) > MAX_REPORT: + raise RuntimeError("run report exceeds 64 KiB") + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd, temporary = tempfile.mkstemp(prefix=".sync-run-", dir=path.parent) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(raw); stream.flush(); os.fsync(stream.fileno()) + os.replace(temporary, path) + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: os.fsync(directory) + finally: os.close(directory) + finally: + try: os.unlink(temporary) + except FileNotFoundError: pass + + +def load_json(path: Path, limit: int) -> dict: + raw = path.read_bytes() + if len(raw) > limit: raise RuntimeError(f"{path.name} exceeds its bound") + value = json.loads(raw) + if not isinstance(value, dict): raise RuntimeError(f"{path.name} must contain an object") + return value + + +def update(state_path: Path, state: dict, phase: str, **fields: object) -> None: + if phase not in PHASES: raise RuntimeError("invalid run phase") + state.update(fields); state["phase"] = phase + state["progress_revision"] = int(state.get("progress_revision", 0)) + 1 + state["updated_at"] = datetime.now(timezone.utc).isoformat() + atomic_write(state_path, state) + + +def validate_result(value: dict, run_id: str) -> dict: + required = {"format", "version", "run_id", "producer_peer_id", "consumer_peer_id", + "inbound_generation_id", "outbound_generation_id", "manifest_sha256", + "result", "sessions_reconciled", "domains", "drafts", "ai_midpoint", "ai_post_sync"} + if set(value) != required or value.get("format") != "trainlog-sync-worker-report" or value.get("version") != 1: + raise RuntimeError("malformed peer worker report") + if value.get("run_id") != run_id or value.get("result") != "completed": + raise RuntimeError("peer worker did not complete the correlated run") + for identity, prefix in ((value["producer_peer_id"], "peer_"), + (value["consumer_peer_id"], "peer_"), + (value["inbound_generation_id"], "gen_"), + (value["outbound_generation_id"], "gen_")): + if not isinstance(identity, str) or not identity.startswith(prefix) or len(identity) > 64: + raise RuntimeError("invalid worker identity") + if not isinstance(value["sessions_reconciled"], int) or value["sessions_reconciled"] < 0: + raise RuntimeError("invalid sessions_reconciled") + if not isinstance(value["domains"], dict) or not all(isinstance(k, str) and isinstance(v, bool) for k, v in value["domains"].items()): + raise RuntimeError("invalid domain coverage") + if not isinstance(value["drafts"], list) or len(value["drafts"]) > 32: + raise RuntimeError("invalid draft summary") + return value + + +def run(args: argparse.Namespace) -> int: + state_path, config_path = Path(args.state), Path(args.config) + state = load_json(state_path, MAX_REPORT) + config = load_json(config_path, MAX_CONFIG) + if state.get("run_id") != args.run_id or state.get("request_id") != args.request_id: + raise RuntimeError("admission identity mismatch") + if set(config) != {"format", "version", "enabled", "mode", "capabilities", "peer_command", "timeout_seconds"}: + raise RuntimeError("invalid trusted sync configuration") + if config["format"] != "trainlog-sync-orchestrator-config" or config["version"] != 1 or config["enabled"] is not True: + raise RuntimeError("full-generation synchronization is disabled") + required_caps = {"generation-manifest-v1", "generation-ack-v1", "causal-delete-v1", "mobile-history-v4", "execution-draft-v1"} + caps = config["capabilities"] + if not isinstance(caps, list) or not required_caps.issubset(set(caps)): + update(state_path, state, "explicitly_degraded", result="incompatible", + missing_capabilities=sorted(required_caps - set(caps if isinstance(caps, list) else [])), + diagnostic="Peer lacks mandatory full-generation capabilities", finished_at=datetime.now(timezone.utc).isoformat()) + return 3 + command = config["peer_command"] + if not isinstance(command, list) or not command or len(command) > 16 or not all(isinstance(part, str) and 0 < len(part) <= 4096 for part in command): + raise RuntimeError("invalid trusted peer command") + timeout = config["timeout_seconds"] + if not isinstance(timeout, int) or timeout < 1 or timeout > 900: + raise RuntimeError("invalid worker timeout") + # CONTRACT: this is the same XDG data-directory lock used by the legacy + # C engine, so Web, TUI and daemon admissions cannot overlap. + lock_path = Path(args.database).parent / "sync.lock" + lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) + try: + try: fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + update(state_path, state, "failed", result="conflict", diagnostic="Another synchronization owns the common lock", finished_at=datetime.now(timezone.utc).isoformat()) + return 4 + update(state_path, state, "waiting_android_publication", result="running", diagnostic="") + update(state_path, state, "running") + environment = os.environ.copy() + environment.update({"TRAINLOG_SYNC_RUN_ID": args.run_id, + "TRAINLOG_SYNC_TRIGGER": state["trigger"], + "TRAINLOG_SYNC_DATABASE": args.database}) + child = subprocess.Popen(command, stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=environment, + start_new_session=True) + try: + stdout, stderr = child.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + os.killpg(child.pid, signal.SIGTERM) + try: stdout, stderr = child.communicate(timeout=5) + except subprocess.TimeoutExpired: + os.killpg(child.pid, signal.SIGKILL); stdout, stderr = child.communicate() + raise RuntimeError("peer worker timed out") + if len(stdout) > MAX_REPORT or len(stderr) > MAX_REPORT: + raise RuntimeError("peer worker output exceeds 64 KiB") + if child.returncode != 0: + detail = stderr.decode("utf-8", "replace").strip()[:MAX_DIAGNOSTIC] + raise RuntimeError(f"peer worker exited {child.returncode}: {detail}") + result = validate_result(json.loads(stdout), args.run_id) + update(state_path, state, "local_import_committed", + inbound_generation_id=result["inbound_generation_id"], + producer_peer_id=result["producer_peer_id"], + consumer_peer_id=result["consumer_peer_id"], + sessions_reconciled=result["sessions_reconciled"], domains=result["domains"], drafts=result["drafts"]) + update(state_path, state, "published", outbound_generation_id=result["outbound_generation_id"]) + update(state_path, state, "waiting_acknowledgement") + update(state_path, state, "peer_consumed", manifest_sha256=result["manifest_sha256"]) + update(state_path, state, "completed", result="completed", diagnostic="", + ai_midpoint=result["ai_midpoint"], ai_post_sync=result["ai_post_sync"], + finished_at=datetime.now(timezone.utc).isoformat()) + return 0 + finally: + os.close(lock_fd) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--database", required=True); parser.add_argument("--state", required=True) + parser.add_argument("--config", required=True); parser.add_argument("--run-id", required=True) + parser.add_argument("--request-id", required=True) + args = parser.parse_args() + try: return run(args) + except (OSError, ValueError, json.JSONDecodeError, RuntimeError, subprocess.TimeoutExpired) as error: + try: + path = Path(args.state); state = load_json(path, MAX_REPORT) + update(path, state, "failed", result="failed", diagnostic=str(error)[:MAX_DIAGNOSTIC], + finished_at=datetime.now(timezone.utc).isoformat()) + except Exception: pass + return 2 + + +if __name__ == "__main__": raise SystemExit(main()) diff --git a/tools/validate_sync_isolated.py b/tools/validate_sync_isolated.py index e2abd0f..7571f92 100755 --- a/tools/validate_sync_isolated.py +++ b/tools/validate_sync_isolated.py @@ -375,7 +375,7 @@ def commands_for_suite(repository: Path, suite: str) -> list[tuple[str, list[str return smoke if suite == "preflight": return [] - return [ + commands = [ ("desktop-compile", ["meson", "compile", "-C", "build"], repository), ( "desktop-tests", @@ -423,6 +423,11 @@ def commands_for_suite(repository: Path, suite: str) -> list[tuple[str, list[str android, ), ] + browser_python = os.environ.get("TRAINLOG_WEB_E2E_PYTHON") + if browser_python: + commands.append(("web-sync-browser", [browser_python, + str(repository / "tests/test_web_sync_browser.py"), "-v"], repository)) + return commands def write_summary(paths: RunPaths, payload: dict[str, object]) -> None: diff --git a/tui/include/trainlog/web_server.h b/tui/include/trainlog/web_server.h index efef344..9f38792 100644 --- a/tui/include/trainlog/web_server.h +++ b/tui/include/trainlog/web_server.h @@ -10,7 +10,8 @@ /* Blocking, single-threaded local HTTP adapter. Port zero is reserved for * controlled test harnesses and is rejected by the CLI. The database remains * borrowed and process-owned; this infrastructure route never reads it. */ -int trainlog_web_server_run(TrainlogDatabase *database, uint16_t port, +int trainlog_web_server_run(TrainlogDatabase *database, const char *database_path, + uint16_t port, const volatile sig_atomic_t *stop_requested, char *diagnostic, size_t diagnostic_capacity); diff --git a/tui/meson.build b/tui/meson.build index 69b548e..442a1b6 100644 --- a/tui/meson.build +++ b/tui/meson.build @@ -47,12 +47,16 @@ if web_assets_enabled '../web/src/api/health.ts', '../web/src/api/dashboard.ts', '../web/src/api/dashboardLayout.ts', + '../web/src/api/sync.ts', + '../web/src/api/sync.test.ts', '../web/src/app/App.tsx', '../web/src/app/routes.ts', '../web/src/app/useRoute.ts', '../web/src/components/EmptyState.tsx', '../web/src/components/Footer.tsx', '../web/src/components/Header.tsx', + '../web/src/components/SyncControl.tsx', + '../web/src/components/SyncControl.test.tsx', '../web/src/components/Tile.tsx', '../web/src/charts/ProgressionChart.tsx', '../web/src/charts/trainlogChartTheme.ts', @@ -730,6 +734,11 @@ test( args: [meson.project_source_root() / 'tests/test_sync_generation_exchange.py'], timeout: 120, ) +test( + 'sync_orchestrator', + find_program('python3'), + args: [meson.project_source_root() / 'tests/test_sync_orchestrator.py'], +) test( 'schema_v21_migration', executable('test_schema_v21_migration', 'tests/test_schema_v21_migration.c', dependencies: trainlog_core_dep), diff --git a/tui/src/main.c b/tui/src/main.c index 5c4faec..5e98d95 100644 --- a/tui/src/main.c +++ b/tui/src/main.c @@ -20,7 +20,8 @@ static void request_web_stop(int signal_number) web_stop_requested = 1; } -static int run_web(TrainlogDatabase *database, uint16_t port) +static int run_web(TrainlogDatabase *database, const char *database_path, + uint16_t port) { struct sigaction action; struct sigaction previous_interrupt; @@ -42,7 +43,7 @@ static int run_web(TrainlogDatabase *database, uint16_t port) "trainlog: impossible d'installer les gestionnaires de signaux\n"); return 1; } - result = trainlog_web_server_run(database, port, &web_stop_requested, + result = trainlog_web_server_run(database, database_path, port, &web_stop_requested, diagnostic, sizeof(diagnostic)); (void)sigaction(SIGINT, &previous_interrupt, NULL); (void)sigaction(SIGTERM, &previous_terminate, NULL); @@ -105,7 +106,7 @@ int main(int argc, char *argv[]) return 1; } result = options.mode == TRAINLOG_RUN_WEB - ? run_web(database, options.port) + ? run_web(database, database_path, options.port) : trainlog_tui_run(database); trainlog_database_close(database); return result; diff --git a/tui/src/web_server.c b/tui/src/web_server.c index 9ffb36a..c0dac14 100644 --- a/tui/src/web_server.c +++ b/tui/src/web_server.c @@ -2,6 +2,8 @@ #include #include +#include +#include #include #include #include @@ -10,7 +12,13 @@ #include #include #include +#include +#include +#include #include +#include +#include +#include #include "trainlog/web_dashboard.h" #include "trainlog/dashboard_layout.h" @@ -28,6 +36,9 @@ typedef struct TrainlogWebContext { uint16_t port; TrainlogDatabase *database; + const char *database_path; + char state_path[PATH_MAX + 32U]; + pid_t worker_pid; char csrf_token[65]; bool invalid_layout_reported; } TrainlogWebContext; @@ -105,6 +116,147 @@ static enum MHD_Result queue_json(struct MHD_Connection *connection, return result; } +static enum MHD_Result queue_sync_json(struct MHD_Connection *connection, + unsigned int status, const char *body, const char *csrf_token, + const char *location) +{ + struct MHD_Response *response = MHD_create_response_from_buffer(strlen(body), + (void *)body, MHD_RESPMEM_MUST_COPY); + enum MHD_Result result; + if (response == NULL) return MHD_NO; + if (MHD_add_response_header(response, MHD_HTTP_HEADER_CONTENT_TYPE, + "application/json; charset=utf-8") != MHD_YES || + MHD_add_response_header(response, MHD_HTTP_HEADER_CACHE_CONTROL, + "no-store") != MHD_YES || + MHD_add_response_header(response, + MHD_HTTP_HEADER_X_CONTENT_TYPE_OPTIONS, "nosniff") != MHD_YES || + MHD_add_response_header(response, "Content-Security-Policy", + "default-src 'none'; frame-ancestors 'none'; base-uri 'none'") != MHD_YES || + MHD_add_response_header(response, "X-Trainlog-CSRF-Token", csrf_token) != MHD_YES || + (location != NULL && MHD_add_response_header(response, + MHD_HTTP_HEADER_LOCATION, location) != MHD_YES)) { + MHD_destroy_response(response); return MHD_NO; + } + result = MHD_queue_response(connection, status, response); + MHD_destroy_response(response); return result; +} + +static bool read_bounded_file(const char *path, char *output, size_t capacity) +{ + int descriptor; struct stat info; size_t used = 0U; + if (path == NULL || output == NULL || capacity < 2U) return false; + descriptor = open(path, O_RDONLY | O_CLOEXEC | O_NOFOLLOW); + if (descriptor < 0) return false; + if (fstat(descriptor, &info) != 0 || !S_ISREG(info.st_mode) || + info.st_size < 0 || (uintmax_t)info.st_size >= (uintmax_t)capacity) { + (void)close(descriptor); return false; + } + while (used < (size_t)info.st_size) { + ssize_t count = read(descriptor, output + used, (size_t)info.st_size - used); + if (count < 0 && errno == EINTR) continue; + if (count <= 0) { (void)close(descriptor); return false; } + used += (size_t)count; + } + output[used] = '\0'; (void)close(descriptor); return true; +} + +static bool sync_state_active(const char *json, char request_id[64]) +{ + yyjson_doc *document = yyjson_read(json, strlen(json), 0U); + yyjson_val *root; yyjson_val *phase; yyjson_val *request; + const char *text; + bool active = false; + if (document == NULL) return false; + root = yyjson_doc_get_root(document); phase = yyjson_obj_get(root, "phase"); + request = yyjson_obj_get(root, "request_id"); + if (yyjson_is_str(request) && strlen(yyjson_get_str(request)) < 64U) + (void)snprintf(request_id, 64U, "%s", yyjson_get_str(request)); + if (yyjson_is_str(phase)) { + text = yyjson_get_str(phase); + active = strcmp(text, "requested") == 0 || + strcmp(text, "waiting_android_publication") == 0 || + strcmp(text, "running") == 0 || + strcmp(text, "local_import_committed") == 0 || + strcmp(text, "published") == 0 || + strcmp(text, "waiting_acknowledgement") == 0 || + strcmp(text, "peer_consumed") == 0; + } + yyjson_doc_free(document); return active; +} + +static bool valid_sync_request(const char *body, size_t size, char request_id[64]) +{ + yyjson_doc *document = yyjson_read(body, size, 0U); + yyjson_val *root; yyjson_val *request; yyjson_val *trigger; + const char *id; + if (document == NULL) return false; + root = yyjson_doc_get_root(document); + if (!yyjson_is_obj(root) || yyjson_obj_size(root) != 2U) { + yyjson_doc_free(document); return false; + } + request = yyjson_obj_get(root, "request_id"); trigger = yyjson_obj_get(root, "trigger"); + if (!yyjson_is_str(request) || !yyjson_is_str(trigger) || + strcmp(yyjson_get_str(trigger), "web") != 0) { + yyjson_doc_free(document); return false; + } + id = yyjson_get_str(request); + if (strlen(id) != 39U || strncmp(id, "sy_", 3U) != 0 || + uuid_parse(id + 3U, (unsigned char[16]){0}) != 0) { + yyjson_doc_free(document); return false; + } + (void)snprintf(request_id, 64U, "%s", id); + yyjson_doc_free(document); return true; +} + +static bool write_initial_sync_state(const TrainlogWebContext *context, + const char *request_id, char run_id[64], char output[2048]) +{ + uuid_t identifier; char uuid[37]; char temporary[PATH_MAX + 64U]; + char timestamp[40]; time_t now = time(NULL); struct tm broken_down; + int descriptor; int written; ssize_t count; + uuid_generate_random(identifier); uuid_unparse_lower(identifier, uuid); + (void)snprintf(run_id, 64U, "sy_%s", uuid); + if (now == (time_t)-1 || gmtime_r(&now, &broken_down) == NULL || + strftime(timestamp, sizeof(timestamp), "%Y-%m-%dT%H:%M:%SZ", &broken_down) == 0U) + return false; + written = snprintf(output, 2048U, + "{\"ai_midpoint\":{\"result\":\"not_configured\"},\"ai_post_sync\":{\"result\":\"not_configured\"}," + "\"diagnostic\":\"\",\"domains\":{},\"drafts\":[],\"effective_mode\":\"full_generation_v1\"," + "\"finished_at\":null,\"inbound_generation_id\":null,\"missing_capabilities\":[]," + "\"outbound_generation_id\":null,\"phase\":\"requested\",\"producer_peer_id\":null," + "\"consumer_peer_id\":null,\"progress_revision\":1,\"request_id\":\"%s\"," + "\"requested_mode\":\"full_generation_v1\",\"result\":\"running\",\"run_id\":\"%s\"," + "\"sessions_reconciled\":null,\"started_at\":\"%s\",\"trigger\":\"web\",\"updated_at\":\"%s\"}\n", + request_id, run_id, timestamp, timestamp); + if (written < 0 || written >= 2048) return false; + if (snprintf(temporary, sizeof(temporary), "%s.new.%ld", context->state_path, + (long)getpid()) < 0) return false; + descriptor = open(temporary, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC, 0600); + if (descriptor < 0) return false; + count = write(descriptor, output, (size_t)written); + if (count != written || fsync(descriptor) != 0 || close(descriptor) != 0 || + rename(temporary, context->state_path) != 0) { + (void)close(descriptor); (void)unlink(temporary); return false; + } + return true; +} + +static bool launch_sync_worker(TrainlogWebContext *context, const char *config, + const char *run_id, const char *request_id) +{ + pid_t child = fork(); + if (child < 0) return false; + if (child == 0) { + (void)setpgid(0, 0); + execlp("python3", "python3", TRAINLOG_TOOLS_DIR "/sync_orchestrator.py", + "--database", context->database_path, "--state", context->state_path, + "--config", config, "--run-id", run_id, "--request-id", request_id, + (char *)NULL); + _exit(127); + } + context->worker_pid = child; return true; +} + static enum MHD_Result queue_asset(struct MHD_Connection *connection, const TrainlogWebAsset *asset) { @@ -275,6 +427,72 @@ static enum MHD_Result handle_request(void *closure, "{\"error\":\"method_not_allowed\"}\n", "GET"); return queue_json(connection, MHD_HTTP_OK, HEALTH, NULL); } + if (strcmp(url, "/api/v1/sync/status") == 0) { + static const char IDLE_DISABLED[] = + "{\"api_version\":1,\"enabled\":false,\"phase\":\"idle\"," + "\"result\":\"disabled\",\"last_success\":null}\n"; + static const char IDLE_ENABLED[] = + "{\"api_version\":1,\"enabled\":true,\"phase\":\"idle\"," + "\"result\":\"idle\",\"last_success\":null}\n"; + char status_json[65537]; + if (!is_get) + return queue_json(connection, MHD_HTTP_METHOD_NOT_ALLOWED, + "{\"error\":\"method_not_allowed\"}\n", "GET"); + if (!read_bounded_file(context->state_path, status_json, + sizeof(status_json))) + return queue_sync_json(connection, MHD_HTTP_OK, + getenv("TRAINLOG_SYNC_GENERATION_CONFIG") == NULL + ? IDLE_DISABLED : IDLE_ENABLED, + context->csrf_token, NULL); + return queue_sync_json(connection, MHD_HTTP_OK, status_json, + context->csrf_token, NULL); + } + if (strcmp(url, "/api/v1/sync") == 0) { + const char *origin; const char *csrf; const char *content_type; + const char *config = getenv("TRAINLOG_SYNC_GENERATION_CONFIG"); + char request_id[64] = ""; char existing_id[64] = ""; + char existing[65537]; char initial[2048]; char run_id[64]; + char accepted[256]; + if (strcmp(method, MHD_HTTP_METHOD_POST) != 0) + return queue_json(connection, MHD_HTTP_METHOD_NOT_ALLOWED, + "{\"error\":\"method_not_allowed\"}\n", "POST"); + origin = MHD_lookup_connection_value(connection, MHD_HEADER_KIND, "Origin"); + csrf = MHD_lookup_connection_value(connection, MHD_HEADER_KIND, + "X-Trainlog-CSRF-Token"); + if (!valid_origin(context, origin) || !valid_csrf(context, csrf)) + return queue_json(connection, MHD_HTTP_FORBIDDEN, + "{\"error\":\"mutation_forbidden\"}\n", NULL); + content_type = MHD_lookup_connection_value(connection, MHD_HEADER_KIND, + MHD_HTTP_HEADER_CONTENT_TYPE); + if (content_type == NULL || strcmp(content_type, "application/json") != 0) + return queue_json(connection, MHD_HTTP_UNSUPPORTED_MEDIA_TYPE, + "{\"error\":\"unsupported_media_type\"}\n", NULL); + if (!valid_sync_request(state->body, state->body_size, request_id)) + return queue_json(connection, MHD_HTTP_BAD_REQUEST, + "{\"error\":\"invalid_sync_request\"}\n", NULL); + if (config == NULL || config[0] != '/' || strlen(config) >= PATH_MAX) + return queue_json(connection, MHD_HTTP_CONFLICT, + "{\"error\":\"full_generation_disabled\"}\n", NULL); + if (read_bounded_file(context->state_path, existing, sizeof(existing))) { + bool active = sync_state_active(existing, existing_id); + if (strcmp(existing_id, request_id) == 0) + return queue_sync_json(connection, active ? MHD_HTTP_ACCEPTED : MHD_HTTP_OK, + existing, context->csrf_token, "/api/v1/sync/status"); + if (active) + return queue_json(connection, MHD_HTTP_CONFLICT, + "{\"error\":\"sync_already_running\"}\n", NULL); + } + if (!write_initial_sync_state(context, request_id, run_id, initial) || + !launch_sync_worker(context, config, run_id, request_id)) + return queue_json(connection, MHD_HTTP_INTERNAL_SERVER_ERROR, + "{\"error\":\"sync_worker_start_failed\"}\n", NULL); + (void)snprintf(accepted, sizeof(accepted), + "{\"api_version\":1,\"phase\":\"requested\",\"result\":\"running\",\"progress_revision\":1,\"request_id\":\"%s\"," + "\"run_id\":\"%s\",\"status_url\":\"/api/v1/sync/status\"}\n", + request_id, run_id); + return queue_sync_json(connection, MHD_HTTP_ACCEPTED, accepted, + context->csrf_token, "/api/v1/sync/status"); + } if (strcmp(url, "/api/v1/dashboard") == 0) { TrainlogWebDashboardQuery query; TrainlogWebDashboardSnapshot snapshot; @@ -424,7 +642,8 @@ static enum MHD_Result accept_loopback_only(void *closure, return ipv4->sin_addr.s_addr == htonl(INADDR_LOOPBACK) ? MHD_YES : MHD_NO; } -int trainlog_web_server_run(TrainlogDatabase *database, uint16_t port, +int trainlog_web_server_run(TrainlogDatabase *database, const char *database_path, + uint16_t port, const volatile sig_atomic_t *stop_requested, char *diagnostic, size_t diagnostic_capacity) { @@ -432,7 +651,8 @@ int trainlog_web_server_run(TrainlogDatabase *database, uint16_t port, struct MHD_Daemon *daemon; TrainlogWebContext context; int result = 0; - if (database == NULL || stop_requested == NULL) return -1; + if (database == NULL || database_path == NULL || database_path[0] != '/' || + stop_requested == NULL) return -1; web_diagnostic(diagnostic, diagnostic_capacity, ""); if (!trainlog_web_assets_available()) { web_diagnostic(diagnostic, diagnostic_capacity, @@ -445,6 +665,13 @@ int trainlog_web_server_run(TrainlogDatabase *database, uint16_t port, bind_address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); context.port = port; context.database = database; + context.database_path = database_path; + context.worker_pid = -1; + if (snprintf(context.state_path, sizeof(context.state_path), "%s.sync-run.json", + database_path) < 0 || strlen(context.state_path) >= sizeof(context.state_path) - 1U) { + web_diagnostic(diagnostic, diagnostic_capacity, "chemin d'état sync trop long"); + return -1; + } context.invalid_layout_reported = false; if (!generate_csrf_token(context.csrf_token)) { web_diagnostic(diagnostic, diagnostic_capacity, @@ -482,6 +709,12 @@ int trainlog_web_server_run(TrainlogDatabase *database, uint16_t port, unsigned long long timeout_ms = TRAINLOG_HTTP_POLL_MAX_MS; int maximum_fd = 0; int selected; + if (context.worker_pid > 0) { + int worker_status; + if (waitpid(context.worker_pid, &worker_status, WNOHANG) == + context.worker_pid) + context.worker_pid = -1; + } FD_ZERO(&read_set); FD_ZERO(&write_set); FD_ZERO(&except_set); @@ -515,5 +748,23 @@ int trainlog_web_server_run(TrainlogDatabase *database, uint16_t port, } } MHD_stop_daemon(daemon); + if (context.worker_pid > 0) { + int worker_status; + pid_t waited = waitpid(context.worker_pid, &worker_status, WNOHANG); + if (waited == 0) { + size_t attempt; + struct timespec delay = {0, 100000000L}; + (void)kill(-context.worker_pid, SIGTERM); + for (attempt = 0U; attempt < 50U; ++attempt) { + waited = waitpid(context.worker_pid, &worker_status, WNOHANG); + if (waited == context.worker_pid || waited < 0) break; + (void)nanosleep(&delay, NULL); + } + if (waited == 0) { + (void)kill(-context.worker_pid, SIGKILL); + (void)waitpid(context.worker_pid, &worker_status, 0); + } + } + } return result; } diff --git a/tui/tests/test_web_server.c b/tui/tests/test_web_server.c index 5196fb6..9ce7597 100644 --- a/tui/tests/test_web_server.c +++ b/tui/tests/test_web_server.c @@ -145,6 +145,7 @@ static bool test_http_contract(TrainlogDatabase *database) pid_t child; int status; struct sigaction action; + (void)unlink("/tmp/trainlog-web-test.db.sync-run.json"); CHECK(reserve_port(&port, 0) == -2); child = fork(); CHECK(child >= 0); @@ -155,7 +156,7 @@ static bool test_http_contract(TrainlogDatabase *database) (void)sigemptyset(&action.sa_mask); (void)sigaction(SIGTERM, &action, NULL); child_stop = 0; - _exit(trainlog_web_server_run(database, port, &child_stop, + _exit(trainlog_web_server_run(database, "/tmp/trainlog-web-test.db", port, &child_stop, diagnostic, sizeof(diagnostic)) == 0 ? 0 : 1); } CHECK(exchange(port, @@ -169,6 +170,27 @@ static bool test_http_contract(TrainlogDatabase *database) CHECK(strstr(response, "{\"api_version\":1,\"status\":\"ok\",\"product\":\"trainlog\"," "\"version\":\"0.1.2\"}") != NULL); + { + char token[80]; + CHECK(exchange(port, + "GET /api/v1/sync/status HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n", + response, sizeof(response)) && strstr(response, "HTTP/1.1 200") != NULL); + CHECK(strstr(response, "\"phase\":\"idle\"") != NULL); + CHECK(response_header(response, "X-Trainlog-CSRF-Token: ", token, + sizeof(token)) && strlen(token) == 64U); + (void)snprintf(request, sizeof(request), + "POST /api/v1/sync HTTP/1.1\r\nHost: 127.0.0.1\r\n" + "Origin: http://127.0.0.1:%u\r\nContent-Type: application/json\r\n" + "X-Trainlog-CSRF-Token: %s\r\nContent-Length: 72\r\n\r\n" + "{\"request_id\":\"sy_11111111-1111-4111-8111-111111111111\",\"trigger\":\"web\"}", + (unsigned int)port, token); + CHECK(exchange(port, request, response, sizeof(response)) && + strstr(response, "HTTP/1.1 409") != NULL && + strstr(response, "full_generation_disabled") != NULL); + CHECK(exchange(port, + "GET /api/v1/sync HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n", + response, sizeof(response)) && strstr(response, "HTTP/1.1 405") != NULL); + } CHECK(exchange(port, "GET /api/v1/dashboard HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n", response, sizeof(response)) && strstr(response, "HTTP/1.1 200") != NULL); @@ -307,7 +329,7 @@ static bool test_port_in_use(TrainlogDatabase *database) uint16_t port; int occupied = reserve_port(&port, 1); CHECK(occupied >= 0); - CHECK(trainlog_web_server_run(database, port, &stop, diagnostic, + CHECK(trainlog_web_server_run(database, "/tmp/trainlog-web-test.db", port, &stop, diagnostic, sizeof(diagnostic)) != 0); CHECK(strstr(diagnostic, "127.0.0.1") != NULL); (void)close(occupied); @@ -339,7 +361,7 @@ static bool test_dashboard_core_error_translation(void) if (child == 0) { volatile sig_atomic_t stop = 0; char diagnostic[256]; - _exit(trainlog_web_server_run(database, port, &stop, diagnostic, + _exit(trainlog_web_server_run(database, path, port, &stop, diagnostic, sizeof(diagnostic)) == 0 ? 0 : 1); } CHECK(exchange(port, @@ -366,7 +388,7 @@ int main(void) if (!trainlog_web_assets_available()) { volatile sig_atomic_t stop = 0; char diagnostic[256]; - passed = trainlog_web_server_run(database, 8080U, &stop, diagnostic, + passed = trainlog_web_server_run(database, "/tmp/trainlog-web-test.db", 8080U, &stop, diagnostic, sizeof(diagnostic)) != 0 && strstr(diagnostic, "frontend") != NULL; } else { passed = test_http_contract(database) && test_port_in_use(database) &&