diff --git a/CHANGELOG.md b/CHANGELOG.md index 18b72eb..ae73fbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## 1.3.2 + +- Preserve events when the writer queue fills by waiting for space instead of discarding them. +- Use faster gzip compression during play; the finished ZIP keeps its existing compression. +- Allow one record larger than the retained-data budget to drain on its own, and release waiting callers if output fails. +- Serialize capture construction before copying bodies or snapshots so concurrent waiting callbacks do not accumulate capture data outside the queue budget. + ## 1.3.1 - Build the runtime, capture plugin and install ZIP with `py scripts/build.py`; version and source commit are read automatically. diff --git a/README.md b/README.md index cf067d1..de68186 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ Raw bodies keep the original bytes, including any account or session fields in t ## How recording works -The plugin observes the game's HTTP send, response decoder and completion paths. It preserves the original API bodies before decoding, so unknown fields remain available. Combat hooks copy selected native state while it is valid, then queue detached records for background writing and compression. It avoids per-frame polling and keeps its queue bounded; any dropped events are reported. +The plugin observes the game's HTTP send, response decoder and completion paths. It preserves the original API bodies before decoding, so unknown fields remain available. Combat hooks copy selected native state while it is valid, then queue detached records for background writing and compression. It avoids per-frame polling. A full queue waits for the writer instead of discarding events; faster compression during play helps keep it draining. Sustained slow storage can still cause waits. Completed gzip blocks are checkpointed during play. On exit, the writer streams them into a ZIP and calculates the manifest hashes. Recovery uses those checkpoints to exclude an unfinished crash tail. Older recordings without checkpoints are marked uncertain and their temporary files are retained. diff --git a/Sources/AtelierCapture/AtelierCapture.csproj b/Sources/AtelierCapture/AtelierCapture.csproj index ba31ad2..1500881 100644 --- a/Sources/AtelierCapture/AtelierCapture.csproj +++ b/Sources/AtelierCapture/AtelierCapture.csproj @@ -1,7 +1,7 @@ net6.0 - 1.3.1 + 1.3.2 enable enable latest diff --git a/Sources/AtelierCapture/FORMAT.md b/Sources/AtelierCapture/FORMAT.md index a1dca13..c223e62 100644 --- a/Sources/AtelierCapture/FORMAT.md +++ b/Sources/AtelierCapture/FORMAT.md @@ -4,7 +4,7 @@ Schema 4 recordings use one ZIP with five flat entries: `api.jsonl`, `combat.jso `manifest_version` is 1 and `format` is `resleriana-capture`. Each manifest entry describes one extracted JSONL stream: `name`, `sha256`, `bytes` and `records`. Hash the exact extracted bytes, including line endings; `records` counts LF bytes. `manifest.json` does not hash itself. ZIP64 sizes must be supported for long recordings. -The manifest copies `schema`, `run_id`, `environment`, timestamps, stopwatch frequency, raw-data flag and producer `metadata` from the original diagnostic session records. Producer identity is self-reported. `capture_status` lists installation/reader results, skipped hooks, observed combat calls and observer failures. `observer_errors` uses cumulative totals when a final status and session end are available; otherwise it counts available warning records. These totals span both environments in the run. `observer_error_samples` counts sampled warnings separately. `dropped_events` totals writer-queue gaps. +The manifest copies `schema`, `run_id`, `environment`, timestamps, stopwatch frequency, raw-data flag and producer `metadata` from the original diagnostic session records. Producer identity is self-reported. `capture_status` lists installation/reader results, skipped hooks, observed combat calls and observer failures. `observer_errors` uses cumulative totals when a final status and session end are available; otherwise it counts available warning records. These totals span both environments in the run. `observer_error_samples` counts sampled warnings separately. `dropped_events` totals writer-queue gaps reported by older producers. Since 1.3.2, queue saturation waits for space instead of dropping an event. Writer failures still stop recording and mark the archive incomplete. `shutdown_complete` means the writer recorded a normal session end. It does not mean every gameplay action was captured. `recovered`, `checkpoint_verified`, `recovery_uncertain` and `discarded_compressed_bytes` distinguish recovery from normal shutdown. Older partial recordings without checkpoints are uncertain; their original temporary files are kept. Missing or corrupt schema 4 checkpoints prevent packaging, and the originals remain available. @@ -20,7 +20,7 @@ Run/environment metadata must agree across records. Event sequences are unique w Diagnostic markers do not all have the event envelope. `session_start` is written lazily, so event timestamps can precede it. `session_end.events` counts queued envelope records across the four streams and excludes diagnostic markers. Do not infer corruption from timestamp order or from markers without an event sequence. -`raw-bodies.jsonl` preserves original entity bytes as Base64, with byte counts and SHA-256 digests. Unknown fields can be decoded later. Common sensitive fields are redacted in decoded logs, but the original raw bodies are retained. Limits apply to each body/snapshot, not to the length of a play session: bodies over 32 MiB are omitted explicitly, snapshots have a 12,000-node/10-level limit, collections stop at 4,096 items, and the writer queue holds at most 2,048 events/64 MiB. Unawaited async results are marked; hooked downstream consumers provide the resolved battle state. +`raw-bodies.jsonl` preserves original entity bytes as Base64, with byte counts and SHA-256 digests. Unknown fields can be decoded later. Common sensitive fields are redacted in decoded logs, but the original raw bodies are retained. Limits apply to each body/snapshot, not to the length of a play session: bodies over 32 MiB are omitted explicitly, snapshots have a 12,000-node/10-level limit, and collections stop at 4,096 items. The writer queue holds at most 2,048 waiting events with a 64 MiB retained-data budget; one larger record may enter when the budget is otherwise empty. Observer construction is serialized before copying native bodies or building snapshots, so concurrent callbacks do not accumulate detached copies while waiting. The queue budget excludes the current observer's working data and serialization/compression buffers; it is not a total process-memory limit. Completed combat records are serialized as captured, without inferring replacement values. Unawaited async results are marked; hooked downstream consumers provide the resolved battle state. ## Trust boundary diff --git a/Sources/AtelierCapture/Plugin.cs b/Sources/AtelierCapture/Plugin.cs index ad6625c..da517df 100644 --- a/Sources/AtelierCapture/Plugin.cs +++ b/Sources/AtelierCapture/Plugin.cs @@ -16,6 +16,7 @@ public sealed class Plugin : BasePlugin internal static new ManualLogSource Log = null!; internal static SessionWriter Writer = null!; internal static Privacy Privacy = null!; + private static readonly object ObservationGate = new(); private static readonly HashSet ObserverWarnings = new(); private static readonly ConcurrentDictionary ObserverErrors = new(); private static readonly ConcurrentDictionary HookStatus = new(); @@ -111,8 +112,11 @@ internal static void Patch(Harmony harmony, MethodBase? target, Type owner, stri internal static void Observe(string kind, Action action) { - try { action(); } - catch (Exception ex) { ReportFailure(kind, ex); } + lock (ObservationGate) + { + try { action(); } + catch (Exception ex) { ReportFailure(kind, ex); } + } } internal static void ReportFailure(string kind, Exception ex) diff --git a/Sources/AtelierCapture/SessionWriter.cs b/Sources/AtelierCapture/SessionWriter.cs index 7dd61e6..b5e1616 100644 --- a/Sources/AtelierCapture/SessionWriter.cs +++ b/Sources/AtelierCapture/SessionWriter.cs @@ -11,6 +11,8 @@ internal sealed class SessionWriter : IDisposable { internal const long QueueBudget = 64 * 1024 * 1024; private readonly BlockingCollection queue = new(2048); + private readonly object admission = new(); + private readonly CancellationTokenSource writerStopped = new(); private readonly Source[] sources = { new("Official"), new("Private") }; private static readonly string[] FileNames = { "api.jsonl.gz", "combat.jsonl.gz", "raw-bodies.jsonl.gz", "diagnostics.jsonl.gz" }; private static readonly string[] StreamNames = { "api", "combat", "raw", "diagnostics" }; @@ -64,19 +66,20 @@ record = line; retainedBytes = line.Length * 2L + 256; } var size = Math.Max(4096, retainedBytes); - if (Interlocked.Add(ref queuedBytes, size) > QueueBudget) + lock (admission) { - Interlocked.Add(ref queuedBytes, -size); - Interlocked.Increment(ref source.Dropped); - return; + // One oversized body may enter an empty queue; it must not be discarded. + while (queuedBytes != 0 && size > QueueBudget - queuedBytes && Volatile.Read(ref failed) == 0) + Monitor.Wait(admission); + if (Volatile.Read(ref failed) != 0) return; + queuedBytes += size; } try { - if (queue.TryAdd(new Entry(source, stream, record, size))) return; + queue.Add(new Entry(source, stream, record, size), writerStopped.Token); } - catch (InvalidOperationException) { } - Interlocked.Add(ref queuedBytes, -size); - Interlocked.Increment(ref source.Dropped); + catch (OperationCanceledException) { Release(size); } + catch (InvalidOperationException) { Release(size); } } finally { @@ -85,6 +88,15 @@ record = line; } } + private void Release(long size) + { + lock (admission) + { + queuedBytes -= size; + Monitor.PulseAll(admission); + } + } + private void CompleteQueue() { if (Volatile.Read(ref emitting) == 0 && Interlocked.CompareExchange(ref stopping, 2, 1) == 1) @@ -101,10 +113,9 @@ private void WriteLoop() if (queue.TryTake(out var entry, 250)) { EnsureFiles(entry.Source); - ReportDrops(entry.Source); entry.Source.Logs![entry.Stream].WriteLine( entry.Record as string ?? JsonSerializer.Serialize(entry.Record)); - Interlocked.Add(ref queuedBytes, -entry.Bytes); + Release(entry.Bytes); entry.Source.Events++; } if ((Stopwatch.GetTimestamp() - lastFlush) * 1000.0 / Stopwatch.Frequency >= 1000) @@ -113,10 +124,8 @@ private void WriteLoop() lastFlush = Stopwatch.GetTimestamp(); } } - foreach (var source in sources.Where(s => s.Logs is not null || s.Dropped != 0)) + foreach (var source in sources.Where(s => s.Logs is not null)) { - EnsureFiles(source); - ReportDrops(source); WriteMetadata(source, new { kind = "capture_status", utc = DateTimeOffset.UtcNow, data = status(source.Name == "Private") }); WriteMetadata(source, new { kind = "session_end", run_id = runId, utc = DateTimeOffset.UtcNow, events = source.Events }); } @@ -128,6 +137,8 @@ private void WriteLoop() } finally { + writerStopped.Cancel(); + lock (admission) Monitor.PulseAll(admission); foreach (var source in sources) { var packed = false; @@ -185,17 +196,10 @@ private void EnsureFiles(Source source) private static void WriteMetadata(Source source, object data) => source.Logs![3].WriteLine(JsonSerializer.Serialize(data)); - private static void ReportDrops(Source source) - { - var count = Interlocked.Exchange(ref source.Dropped, 0); - if (count != 0) WriteMetadata(source, new { kind = "capture_gap", utc = DateTimeOffset.UtcNow, dropped_events = count, reason = "writer_queue_full" }); - } - private void Flush() { foreach (var source in sources.Where(s => s.Logs is not null)) { - ReportDrops(source); foreach (var log in source.Logs!) log.Flush(); SaveCheckpoint(source); } @@ -262,7 +266,7 @@ private sealed class Source(string name) internal CompressedLog[]? Logs; internal FileStream? Lease; internal string? Folder; - internal long Dropped, Events; + internal long Events; } private sealed record Entry(Source Source, int Stream, object Record, long Bytes); @@ -279,7 +283,7 @@ private sealed class CompressedLog(string path) : IDisposable internal void WriteLine(string line) { - writer ??= new StreamWriter(new GZipStream(block, CompressionLevel.Optimal, leaveOpen: true), new UTF8Encoding(false), 65536); + writer ??= new StreamWriter(new GZipStream(block, CompressionLevel.Fastest, leaveOpen: true), new UTF8Encoding(false), 65536); writer.WriteLine(line); characters += line.Length; if (characters >= 4 * 1024 * 1024) Flush(); diff --git a/VERSION b/VERSION index 3a3cd8c..1892b92 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.3.1 +1.3.2