From 1cc60267f9a7c3cc0313cbf28afead3a415d832c Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Sun, 27 Sep 2026 10:10:33 +0200 Subject: [PATCH] feat(scheduler): atomic JSON job store jobs.json is { "version": 1, "jobs": [...] } in VolcanicJson camelCase, written to jobs.json..tmp and moved into place through one semaphore. A corrupt file is renamed jobs.json.corrupt- and loading starts empty. Each entry is deserialised on its own, so an entry with an unknown kind or a missing required field is dropped with a warning like any other invalid job instead of setting the whole file aside. --- .../JobStore.cs | 99 +++++++++++++++ .../Scheduler/JobStoreTests.cs | 114 ++++++++++++++++++ 2 files changed, 213 insertions(+) create mode 100644 extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs create mode 100644 tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs diff --git a/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs new file mode 100644 index 0000000..8733582 --- /dev/null +++ b/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs @@ -0,0 +1,99 @@ +using System.Globalization; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Abstractions.Json; + +namespace Volcanic.AgenticCore.Extensions.Scheduler; + +/// +/// jobs.json: { "version": 1, "jobs": [...] } in camelCase. Written atomically +/// (temp file + rename) through one semaphore. A corrupt file is set aside and loading starts empty, so a bad file never +/// keeps the host down; invalid jobs (including entries that don't deserialise) are dropped with a warning. +/// +public sealed class JobStore(string path, TimeProvider time, ILogger logger) +{ + public const int Version = 1; + private readonly SemaphoreSlim _writeGate = new(1, 1); + + public string Path => path; + + public async Task> LoadAsync(CancellationToken cancellationToken) + { + if (!File.Exists(path)) return []; + var json = await File.ReadAllTextAsync(path, cancellationToken).ConfigureAwait(false); + StoredFile? file; + try + { + file = VolcanicJson.Deserialize(json); + } + catch (JsonException) + { + file = null; + } + if (file is not { Version: Version, Jobs: not null }) + { + var aside = $"{path}.corrupt-{time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}"; + File.Move(path, aside, overwrite: true); + logger.LogError("The job store {Path} could not be read; it was moved to {Aside} and the scheduler starts empty", path, aside); + return []; + } + + var jobs = new List(); + var ids = new HashSet(StringComparer.Ordinal); + for (var i = 0; i < file.Jobs.Count; i++) + { + var (job, problem) = Read(file.Jobs[i]); + problem ??= ids.Add(job!.Id) ? null : "duplicate id"; + if (problem is not null) + { + logger.LogWarning("Dropped scheduled job #{Index} ({Id}) from {Path}: {Problem}", i, job?.Id, path, problem); + continue; + } + jobs.Add(job!); + } + return jobs; + } + + public async Task SaveAsync(IReadOnlyList jobs, CancellationToken cancellationToken) + { + await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false); + var temp = $"{path}.{Environment.ProcessId.ToString(CultureInfo.InvariantCulture)}.tmp"; + try + { + Directory.CreateDirectory(System.IO.Path.GetDirectoryName(System.IO.Path.GetFullPath(path))!); + var json = JsonSerializer.Serialize(new JobFile(Version, [.. jobs]), VolcanicJson.Indented); + await File.WriteAllTextAsync(temp, json, cancellationToken).ConfigureAwait(false); + File.Move(temp, path, overwrite: true); + } + catch + { + try { File.Delete(temp); } catch (IOException) { } catch (UnauthorizedAccessException) { } + throw; + } + finally + { + _writeGate.Release(); + } + } + + /// One entry of the file: the job, or why it's dropped. An entry that doesn't deserialise only loses itself. + private static (ScheduledJob? Job, string? Problem) Read(JsonElement entry) + { + if (entry.ValueKind != JsonValueKind.Object) return (null, "not a JSON object"); + ScheduledJob? job; + try + { + job = VolcanicJson.Deserialize(entry); + } + catch (JsonException ex) + { + return (null, ex.Message); + } + if (job is null) return (null, "empty entry"); + return (job, JobValidation.CheckStored(job)); + } + + private sealed record JobFile(int Version, List Jobs); + + private sealed record StoredFile(int Version, List? Jobs); +} diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs new file mode 100644 index 0000000..a1447b9 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs @@ -0,0 +1,114 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Time.Testing; +using Volcanic.AgenticCore.Extensions.Scheduler; + +namespace Volcanic.AgenticCore.Extensions.Tests.Scheduler; + +public sealed class JobStoreTests : IDisposable +{ + private static readonly DateTimeOffset Now = new(2026, 9, 26, 14, 3, 0, TimeSpan.Zero); + private readonly string _root = Directory.CreateTempSubdirectory("jobstore-").FullName; + private readonly FakeTimeProvider _time = new(Now); + + private string StorePath => Path.Combine(_root, "scheduler", "jobs.json"); + private JobStore Store() => new(StorePath, _time, NullLogger.Instance); + + internal static ScheduledJob Reminder(string id, string target = "111") => new() + { + Id = id, Target = target, Name = "Call the dentist", Kind = JobKind.Reminder, Text = "Call the dentist", + Schedule = new JobSchedule(Now.AddDays(1), null), CreatedAt = Now, + }; + + public void Dispose() + { + try { Directory.Delete(_root, recursive: true); } catch (IOException) { } + } + + [Fact] + public async Task Missing_file_loads_empty() => Assert.Empty(await Store().LoadAsync(CancellationToken.None)); + + [Fact] + public async Task Round_trip_keeps_jobs_and_leaves_no_temp_file() + { + var watcher = new ScheduledJob + { + Id = "w8r4t", Target = "111", Name = "Price check", Kind = JobKind.Watcher, Command = "scripts/price.sh", + TimeoutSeconds = 60, Schedule = new JobSchedule(null, "*/10 * * * *"), CreatedAt = Now, + LastRunAt = Now, LastResult = new JobResult(false, 1, false, "boom", Now), + }; + await Store().SaveAsync([Reminder("k7m2q"), watcher], CancellationToken.None); + + Assert.Equal([Reminder("k7m2q"), watcher], await Store().LoadAsync(CancellationToken.None)); + Assert.Equal(["jobs.json"], Directory.GetFiles(Path.GetDirectoryName(StorePath)!).Select(Path.GetFileName)); + } + + [Fact] + public async Task File_format_is_versioned_camel_case() + { + await Store().SaveAsync([Reminder("k7m2q")], CancellationToken.None); + var json = await File.ReadAllTextAsync(StorePath); + Assert.Contains("\"version\": 1", json); + Assert.Contains("\"jobs\": [", json); + Assert.Contains("\"kind\": \"reminder\"", json); + Assert.NotEqual(0xEF, (await File.ReadAllBytesAsync(StorePath))[0]); // no BOM + } + + [Fact] + public async Task Corrupt_file_is_set_aside_and_loading_starts_empty() + { + Directory.CreateDirectory(Path.GetDirectoryName(StorePath)!); + await File.WriteAllTextAsync(StorePath, "{ not json"); + + Assert.Empty(await Store().LoadAsync(CancellationToken.None)); + Assert.False(File.Exists(StorePath)); + Assert.Equal("{ not json", await File.ReadAllTextAsync(StorePath + ".corrupt-20260926140300")); + } + + [Fact] + public async Task Invalid_entries_are_dropped() + { + Directory.CreateDirectory(Path.GetDirectoryName(StorePath)!); + await File.WriteAllTextAsync(StorePath, """ + { "version": 1, "jobs": [ + { "id": "good1", "target": "111", "name": "ok", "kind": "reminder", "text": "hi", "schedule": { "cron": "0 9 * * *" }, "createdAt": "2026-09-26T14:03:00+00:00" }, + { "id": "both1", "target": "111", "name": "both", "kind": "reminder", "text": "hi", "schedule": { "at": "2026-09-27T09:00:00+00:00", "cron": "0 9 * * *" }, "createdAt": "2026-09-26T14:03:00+00:00" }, + { "id": "nocmd", "target": "111", "name": "watch", "kind": "watcher", "schedule": { "cron": "* * * * *" }, "createdAt": "2026-09-26T14:03:00+00:00" }, + { "id": "good1", "target": "111", "name": "duplicate id", "kind": "reminder", "text": "hi", "schedule": { "cron": "0 9 * * *" }, "createdAt": "2026-09-26T14:03:00+00:00" } + ] } + """); + + var jobs = await Store().LoadAsync(CancellationToken.None); + + Assert.Equal("ok", Assert.Single(jobs).Name); + } + + [Fact] + public async Task An_entry_that_does_not_deserialise_is_dropped_without_setting_the_file_aside() + { + Directory.CreateDirectory(Path.GetDirectoryName(StorePath)!); + await File.WriteAllTextAsync(StorePath, """ + { "version": 1, "jobs": [ + { "id": "good1", "target": "111", "name": "ok", "kind": "reminder", "text": "hi", "schedule": { "cron": "0 9 * * *" }, "createdAt": "2026-09-26T14:03:00+00:00" }, + { "id": "kind1", "target": "111", "name": "bad kind", "kind": "alarm", "text": "hi", "schedule": { "cron": "0 9 * * *" }, "createdAt": "2026-09-26T14:03:00+00:00" }, + { "id": "noat1", "target": "111", "name": "no createdAt", "kind": "reminder", "text": "hi", "schedule": { "cron": "0 9 * * *" } }, + null, + 42 + ] } + """); + + var jobs = await Store().LoadAsync(CancellationToken.None); + + Assert.Equal("good1", Assert.Single(jobs).Id); + Assert.True(File.Exists(StorePath)); + } + + [Fact] + public async Task Concurrent_saves_are_serialised() + { + var store = Store(); + await Task.WhenAll(Enumerable.Range(0, 20).Select(i => Task.Run(() => store.SaveAsync([Reminder($"id{i:000}")], CancellationToken.None)))); + + Assert.Single(await Store().LoadAsync(CancellationToken.None)); + Assert.Empty(Directory.GetFiles(Path.GetDirectoryName(StorePath)!, "*.tmp")); + } +}