admin/Playa
1
0
Fork 0
forked from admin/VolcanicAgenticCore
Code Pull Requests Activity

feat(scheduler): atomic JSON job store

Browse Source 
jobs.json is { "version": 1, "jobs": [...] } in VolcanicJson camelCase,
written to jobs.json.<pid>.tmp and moved into place through one semaphore.
A corrupt file is renamed jobs.json.corrupt-<timestamp> 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.
This commit is contained in:
EugeneTes
2026-09-27 10:10:33 +02:00
parent 8b642885f0
commit 1cc60267f9
2 changed files with 213 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs Normal file
View File

@@ -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;
/// <summary>
/// <c>jobs.json</c>: <c>{ "version": 1, "jobs": [...] }</c> in <see cref="VolcanicJson"/> 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.
/// </summary>
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<IReadOnlyList<ScheduledJob>> LoadAsync(CancellationToken cancellationToken)
{
if (!File.Exists(path)) return [];
var json = await File.ReadAllTextAsync(path, cancellationToken).ConfigureAwait(false);
StoredFile? file;
try
{
file = VolcanicJson.Deserialize<StoredFile>(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<ScheduledJob>();
var ids = new HashSet<string>(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<ScheduledJob> 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();
}
}
/// <summary>One entry of the file: the job, or why it's dropped. An entry that doesn't deserialise only loses itself.</summary>
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<ScheduledJob>(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<ScheduledJob> Jobs);
private sealed record StoredFile(int Version, List<JsonElement>? Jobs);
}

114
tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobStoreTests.cs Normal file
View File

@@ -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"));
}
}