diff --git a/src/MeshWeaver.ContentCollections/ContentImportExtensions.cs b/src/MeshWeaver.ContentCollections/ContentImportExtensions.cs index 409e1293cc..64b07aadb9 100644 --- a/src/MeshWeaver.ContentCollections/ContentImportExtensions.cs +++ b/src/MeshWeaver.ContentCollections/ContentImportExtensions.cs @@ -1,3 +1,4 @@ +using System.Collections.Immutable; using System.Reactive.Linq; using MeshWeaver.Mesh; using MeshWeaver.Mesh.Security; @@ -201,8 +202,15 @@ string FullPath(string rel) if (!request.Mirror) return writes; - var keep = request.Files - .Select(f => FullPath(f.Path)) + // 🚨 issue #2885: the WRITES are chunked so no delivery ever carries a whole asset + // tree, but the PRUNE is not chunkable — it asks "what is under this folder that the + // source no longer carries", and a chunk answering that from its own slice would + // delete the other chunks' files. So a split write names its FULL keep set here, in + // paths, and the prune stays one authoritative pass. Absent (the unsplit case) the + // set is this request's own Files, exactly as before. + var keep = (request.MirrorKeepPaths is { } declaredKeep + ? declaredKeep.Select(NormalizePath) + : request.Files.Select(f => FullPath(f.Path))) .ToHashSet(StringComparer.OrdinalIgnoreCase); // 🚨 issue #435: when the caller declares which paths the SOURCE owns (a GitSync mirror @@ -509,7 +517,86 @@ public SyncContentFilesBuilder SourceOwned(IReadOnlyList? sourceOwnedPat return this; } - /// Post the sync to the target node's hub. Cold — subscribe to run. + /// + /// 🚨 THE BUDGET — issue #2885. How many packaged (base64) bytes of file content one + /// delivery may accumulate before the next file starts a new one. + /// + /// It is DeliveryPayloadBounds.MemoryStreamBlockBytes, the tighter of the two + /// transport ceilings the mesh declares and the one a failure report about a delivery must + /// itself survive — not a number chosen to make a symptom stop. Measured on the base64 form + /// because that is what the message actually weighs: System.Text.Json renders a + /// byte[] as base64 (×4/3), the packaged payload is then held as a UTF-16 string (×2 in + /// bytes), and Utf8JsonWriter.TranscodeAndWriteRawValue rents up to 3 bytes per char to + /// write it — so the transient peak is several times the delivery, once per hop. + /// + /// What this bounds and what it does not. It bounds the ACCUMULATION: a delivery + /// is never larger than this plus the one file that could not be split, so the delivery's size + /// stops being a function of how much content the Space holds. A single file larger than the + /// budget still travels whole — that is a different axis (a file that large belongs behind a + /// content-store handle, not inline) and it is not the defect this closes. + /// + internal const int PayloadBudgetBytes = DeliveryPayloadBounds.MemoryStreamBlockBytes; + + /// + /// What one file costs the packaged payload: its bytes as base64, plus its path. Exact enough + /// to partition against, and it never touches the bytes. + /// + private static long PackagedCost(InlineContentFile file) + => 4L * ((file.Content.Length + 2) / 3) + file.Path.Length; + + /// + /// Splits the accumulated files into deliveries none of which exceeds + /// , except where one file alone does. A file is never split — + /// it is the atom the receiving handler writes — so the guarantee is + /// delivery ≤ budget + largest single file, which is bounded by the CONTENT rather than + /// by how much of it there is. + /// + private ImmutableList> SplitIntoDeliveries() + { + // An empty sync is ONE request, deliberately: with Mirror it is the "the source carries + // nothing here any more" pass, and collapsing it to zero requests would silently skip a + // prune the caller asked for. + if (_files.Count <= 1) + return [_files.ToImmutableList()]; + + var deliveries = ImmutableList.CreateBuilder>(); + var current = ImmutableList.CreateBuilder(); + var accumulated = 0L; + foreach (var file in _files) + { + var cost = PackagedCost(file); + if (current.Count > 0 && accumulated + cost > PayloadBudgetBytes) + { + deliveries.Add(current.ToImmutable()); + current.Clear(); + accumulated = 0L; + } + current.Add(file); + accumulated += cost; + } + deliveries.Add(current.ToImmutable()); + return deliveries.ToImmutable(); + } + + /// + /// Post the sync to the target node's hub. Cold — subscribe to run. + /// + /// 🚨 Issue #2885 — the write is split, and the PRUNE GOES FIRST. A Space's + /// content/** binaries travel inline, so one request per Space made the delivery as + /// large as the Space's asset tree (28,484,421 bytes for AgenticBusiness; 106,070,300 + /// for AgenticEngineering, whose base64 form does not even fit Orleans' + /// MaxMessageBodySize). Serialising that threw OutOfMemoryException on a + /// production pod — twice, on a payload UNDER every transport bound, because the transcode + /// peaks at several times the payload. No bound can fix an allocation that IS the failure, so + /// the producer stops making it. + /// + /// Why the mirror rides the FIRST delivery, not the last. The prune is one + /// authoritative pass measured against the whole set (MirrorKeepPaths), and putting it + /// first makes the split safe under any receiver: were the keep set ever not honoured, the + /// files it would wrongly prune are precisely the ones the FOLLOWING deliveries are about to + /// write, so the operation still converges on the correct folder. Last-position pruning has no + /// such property. + /// public IObservable Post() { // 🚨 A declared identity (WithAccessContext / ImpersonateAsSystem) wins outright: it is a @@ -520,19 +607,59 @@ public IObservable Post() // first stage, inside the System scope) and had all 409 SyncContentFilesRequest posts // (built in its second stage, outside) failed closed for a null AccessContext. var captured = _identity ?? ContentImportExtensions.CaptureCallerContext(_hub); - var request = new SyncContentFilesRequest(_targetCollection, _targetPath, _files.ToArray()) - { - Mirror = _mirror, - SourceOwnedPaths = _sourceOwnedPaths, - }; var address = new Address(_nodePath); + var deliveries = SplitIntoDeliveries(); + var split = deliveries.Count > 1; + // Collection-relative, the form the mirror compares against — and the form + // SourceOwnedPaths already uses. Only computed when the write was actually split. + var keepPaths = split && _mirror + ? _files.Select(f => CombineTargetPath(f.Path)).ToImmutableList() + : null; + + var requests = deliveries.Select((files, index) => + { + // The mirror is ONE pass and it is the FIRST — see the remark on this method. + var prunes = _mirror && index == 0; + return new SyncContentFilesRequest(_targetCollection, _targetPath, files) + { + Mirror = prunes, + // An additive chunk prunes nothing, so it carries neither prune input. + SourceOwnedPaths = prunes ? _sourceOwnedPaths : null, + MirrorKeepPaths = prunes && split ? keepPaths : null, + }; + }); + // Off-router issuing, same reason as ContentImportBuilder.Post: the router must be neither // end of the request/response pair (ROUTER_TRAFFIC); a non-router hub gets itself back. - return ContentImportExtensions.CarryPostIdentity( - Observable.Defer(() => _hub.NodeOperationIssuingHub() + // Concat, never Merge: the deliveries are sequential so the pod never holds more than one + // of them, which is the entire point — and it keeps the prune pass ordered ahead of the + // writes that follow it. + var posted = requests + .Select(request => Observable.Defer(() => _hub.NodeOperationIssuingHub() .Observe(request, o => ContentImportExtensions.ConfigurePost(o, address, captured)) .Select(d => d.Message) - .Take(1)), - _hub, _identity); + .Take(1))) + .Concat() + // Fold the per-delivery answers into the one the caller has always seen. A failure is + // reported as itself and stops the sequence: TakeUntil disposes the Concat, so the + // deliveries behind it are never posted. + .Scan(ImportContentResponse.Ok(0), (total, answer) => answer.Success + ? ImportContentResponse.Ok(total.FilesImported + answer.FilesImported) + : answer) + .TakeUntil(answer => !answer.Success) + .LastAsync(); + + return ContentImportExtensions.CarryPostIdentity(posted, _hub, _identity); + } + + /// + /// A file's path joined onto 's folder — the collection-relative + /// form the handler's mirror compares against (forward slashes, no leading slash). + /// + private string CombineTargetPath(string filePath) + { + var baseDir = (_targetPath ?? string.Empty).Replace('\\', '/').Trim('/'); + var rel = (filePath ?? string.Empty).Replace('\\', '/').TrimStart('/'); + return baseDir.Length == 0 ? rel : $"{baseDir}/{rel}"; } } diff --git a/src/MeshWeaver.Documentation/Data/Architecture/OversizedDeliveryRefusal.md b/src/MeshWeaver.Documentation/Data/Architecture/OversizedDeliveryRefusal.md index fd4add676f..35b18f97d5 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/OversizedDeliveryRefusal.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/OversizedDeliveryRefusal.md @@ -349,11 +349,77 @@ placement of any bound changes that. Note that bounding *concurrency* (as the st already does, `BatchSize = 5`) is **not** that axis and does not help: it limits how many deliveries are in flight, never how big one of them is. -> **This is why #3046 stays open** while #3047, #3048, #3049 and the rest of that burst do not. Its -> leg — `BuildPodHubRoute`'s `IPodHubGrain.Deliver` to a bulk-import hub — has been guarded since -> #2897 and the delivery went through anyway, which means the payload was *under* the bound and the -> allocation multiple is what killed the pod. Everything above narrows that band; only the producer -> closes it. +## The producer that was building it whole (#2885) + +The producer axis stayed open for a day because none of the stacks names a message type, only a +target: `import/{meshHubId}`, the dedicated bulk-import hub. Four incidents converged on it — +#2885 (twice: 2026-08-31 05:52Z and 2026-09-02 03:08Z) and #3046/#3047/#3048 (2026-09-02 +02:59–03:00Z) — and the identification is arithmetic, not inference. + +**The trigger.** `AgenticBusiness/_GitSync` was written at 2026-09-02 02:49:58Z; the burst begins +five minutes later. That is a forced GitSync re-import, which is exactly the event the first +measurement on #2885 said a six-hour quiet window could not rule out. + +**The message.** A GitSync import mirrors a Space's git-committed `content/**` binaries into its +content collection with `SyncContentFilesRequest`, whose `Files` carry the raw bytes **inline** — +deliberately, so binary assets are never round-tripped through a text API. +`ContentAssetMapper.ToContentSyncs` emits **one** `StaticContentSync` per Space (also deliberately: +the mirror is meant to be one authoritative pass over the whole collection), and +`SyncContentFilesBuilder.Post` turned that one group into **one message**. So the delivery's size +was the size of the Space's entire asset tree. + +**The arithmetic.** `Systemorph/MeshWeaver.Education`, the repo `AgenticBusiness` syncs from: + +| Space | `content/**` bytes | as base64 | as a UTF-16 `RawJson` string | transcode rent (≤3 B/char) | +|---|---|---|---|---| +| `AgenticBusiness` | 28,484,421 (10 files) | ~38 MB | ~76 MB | **~114 MB** | +| `AgenticEngineering` | 106,070,300 (12 videos) | **~141 MB** | ~283 MB | ~424 MB | + +The `AgenticBusiness` payload is comfortably **under** `MaxMessageBodySize`, which is why every +bound on this page let it through and #3046 could correctly observe that "the leg is guarded and the +delivery went through anyway". The rent is what failed. `AgenticEngineering` is the other end of the +same defect and is not an OOM at all: its base64 form exceeds the 100 MiB frame limit outright, so +once the producer-side refusals landed that Space's assets **could not sync at all** — a silent, +total failure of a user-visible feature, produced by the same "one delivery per Space" shape. + +**The fix — chunk the writes, keep the prune one pass.** `SyncContentFilesBuilder.Post` now +partitions the files into deliveries against a budget of +`DeliveryPayloadBounds.MemoryStreamBlockBytes` — the *tighter* of the two transport ceilings this +page already names, measured on the base64 form because that is what the message weighs — and posts +them with `Concat`, so the pod holds one at a time. A file is never split: it is the atom the +receiving handler writes, so the guarantee is `delivery ≤ budget + largest single file`, i.e. the +delivery's size stops being a function of *how much* content the Space holds. + +The prune is the part that does not chunk. It asks "what is under this folder that the source no +longer carries", and a chunk answering that from its own slice would delete the other chunks' files. +So `SyncContentFilesRequest` gains `MirrorKeepPaths`: the full keep set as **paths**, which is all +the prune ever needed and which costs nothing. Exactly one delivery mirrors — **the first**, not the +last. That ordering is the safety property: were the keep set ever not honoured by the receiver, the +files it would wrongly prune are precisely the ones the *following* deliveries are about to write, +so the operation still converges on the correct folder. Last-position pruning has no such property. + +An unsplit sync is byte-for-byte what it was: one request, `MirrorKeepPaths` null, the mirror +measured against its own `Files`. + +### What this still does not cover + +- **A single file larger than the budget travels whole.** The largest asset in the Education repo is + a 9.9 MB video → ~13 MB base64 → ~40 MB rent, which is survivable but is not *bounded* by + anything. A file that large belongs behind a content-store handle rather than inline. That is a + different defect from "the producer built the tree whole" and is deliberately not conflated with it. +- **`DeliveryFailure`'s strip does not fire on a typed payload.** `DeliveryPayloadBounds.IsOversized` + returns false for anything that is not `RawJson` — by design, since guessing at the size of a CLR + object would mean serialising it on the hot path. But a NACK is very often built on the + **receiving** hub, where the message has already been deserialised into its CLR type, so the + construction invariant introduced in #3044 silently does not apply there and the body is echoed + back to the sender verbatim. On the content-sync path the producer fix makes that echo at most one + chunk; as a general property of the invariant it remains, and it is worth knowing before relying on + the strip. +- **Other unbounded inbound bodies at `import/{id}`** are smaller but real: `CreateNodesResponse` + echoes up to 25 full `MeshNode`s the caller reads only `Path` off; `CreateOrUpdateNodeResponse` + echoes the node; `DeleteNodeResponse` carries every deleted path; the import/content manifests are + one entry per node and per content file. All are text and grow with the partition rather than with + its binaries. ## Rules @@ -380,6 +446,18 @@ are in flight, never how big one of them is. - **A `[PreventLogging]`-style attribute cannot reach a type with its own converter.** The resolver strips *properties*; a custom-converted type has none. Check that a redaction attribute is actually read before trusting it. +- **A stack that names only a target address is still identifiable — by arithmetic.** Ask what the + producer at that address builds, measure the SOURCE (the git tree, the collection, the partition), + and multiply through base64 → UTF-16 → transcode rent. #2885 sat open across four incidents for + want of a payload size that no stack carried and that a `du` of the source repository supplied in + one command. +- **When a write is split, the PRUNE is not.** Chunking an operation that also deletes means the + delete must be handed the whole set — as identifiers, never as content — and must run as one pass. + Put that pass FIRST: any over-prune is then repaired by the writes queued behind it. +- **A refusal that lands on an unfixed producer converts an OOM into a silent total failure.** Once + `AgenticEngineering`'s 141 MB sync exceeded the frame limit, the bound did its job and the assets + simply never synced. A bound is a diagnosis, not a remedy: check what the refused producer does + next. ## See also diff --git a/src/MeshWeaver.Documentation/Data/Architecture/StaticRepoImport.md b/src/MeshWeaver.Documentation/Data/Architecture/StaticRepoImport.md index 462413410d..3787f5721d 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/StaticRepoImport.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/StaticRepoImport.md @@ -208,6 +208,12 @@ Two things make that work end to end, and both are easy to break again: Test: `PackageContentAssetInstallTest` — the bytes land at the exact layout `/api/content/{root}/{file}` resolves to, a text asset still round-trips, and an untracked upload survives a re-install. +### 🚨 One sync is MANY deliveries — the bytes are never sent whole (#2885) + +`ContentAssetMapper.ToContentSyncs` still emits **one `StaticContentSync` per Space** — the mirror has to be one authoritative pass over the whole collection — but `SyncContentFilesBuilder.Post` no longer turns that into one message. It partitions the files against `DeliveryPayloadBounds.MemoryStreamBlockBytes` (measured on the base64 form, which is what the message weighs) and posts them with `Concat`, so the portal ever holds one batch. A whole-Space message was 28 MB of course video for `AgenticBusiness` — ~114 MB of transient allocation per hop, which took a production pod down twice — and 141 MB packaged for `AgenticEngineering`, over the frame limit entirely, so that Space's assets silently stopped syncing once the producer-side refusals landed. + +The prune does not chunk: it rides the **first** delivery and carries `SyncContentFilesRequest.MirrorKeepPaths`, the full keep set as paths. First rather than last is the safety property — an over-prune would be repaired by the writes queued behind it. Full argument and arithmetic: [Oversized Delivery Refusal](/Doc/Architecture/OversizedDeliveryRefusal) → "The producer that was building it whole". Tests: `ContentSyncIsNeverBuiltWholeTest`, `ContentSyncMirrorSurvivesTheSplitTest`. + ## The two primitives ### 1. Source fingerprint — the content-version diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-02-course-videos-and-images-sync-again.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-02-course-videos-and-images-sync-again.md new file mode 100644 index 0000000000..a0da23d15c --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-02-course-videos-and-images-sync-again.md @@ -0,0 +1,42 @@ +--- +Name: Course videos and images sync again +Category: Fix +Description: A space whose repository carried more than a few megabytes of videos, images or other files could no longer sync them — the sync was sent as one enormous message that either exhausted a portal's memory or was rejected outright. Files are now written in batches sized to what the platform can carry. +Icon: ArrowSync +Order: -20260902 +--- + +# Course videos and images sync again + +When a space is synced from a git repository, the files committed under its `content/` folder — +course videos, posters, images, fonts, anything that is not text — are copied into the space so the +portal can serve them. That copy was sent as **one message carrying every file in the space at +once**. + +For a space with a handful of small images, that worked and nobody noticed. For a course with video +in it, it did not. One space in the education repository holds 28 MB of video across ten files; +another holds 106 MB across twelve. Packaged for transport, those become roughly 38 MB and 141 MB — +and a message is copied and re-encoded several times on its way, each copy costing several times +its own size again. + +The two ends of that produced two different failures, both silent from the outside: + +- **The 28 MB space exhausted the portal's memory** while packaging the message. This happened on + a live portal on 31 August and again on 2 September. The sync failed; nothing said so. +- **The 141 MB space was simply too large to send at all.** Recent work added a check that refuses + a message bigger than the transport can carry — correctly — so that space's videos stopped + syncing entirely rather than taking a portal down. Also silently. + +**What changed.** The sync no longer builds one message. It writes the files in batches sized to +what the platform can comfortably carry, one batch at a time, so the cost of a sync no longer grows +with the amount of content a space holds. A single very large file still travels on its own, which +is far cheaper than travelling with everything else. + +The part that deletes files the repository no longer carries still happens exactly once, and still +measures the folder against the **complete** set of files being synced — so nothing is deleted +because it happened to travel in a different batch, and, as before, a file you uploaded yourself +that the repository never tracked is left alone. + +What you should notice: syncing a space with videos or large images in it now works, and a big +sync no longer competes with everything else the portal is doing. Spaces small enough to have been +working all along behave exactly as they did. diff --git a/src/MeshWeaver.Mesh.Contract/ImportDeleteRequests.cs b/src/MeshWeaver.Mesh.Contract/ImportDeleteRequests.cs index 4574ff3d72..23250f3bf2 100644 --- a/src/MeshWeaver.Mesh.Contract/ImportDeleteRequests.cs +++ b/src/MeshWeaver.Mesh.Contract/ImportDeleteRequests.cs @@ -148,6 +148,27 @@ public record SyncContentFilesRequest( /// . /// public IReadOnlyList? SourceOwnedPaths { get; init; } + + /// + /// The FULL set of collection-relative paths this mirror must keep, when the write it belongs to + /// was SPLIT across several deliveries. When null (the ordinary, unsplit case) the mirror is + /// measured against , exactly as it always was. + /// + /// 🚨 Issue #2885 — a write is split, a mirror is not. A Space's + /// content/** binaries travel INLINE, so one request per Space made the delivery as large + /// as the Space's asset tree: 28,484,421 bytes of course video for AgenticBusiness — + /// ×4/3 as base64, held as a UTF-16 string, then transcoded through a rent of up to 3 bytes per + /// char — which is the allocation that threw OutOfMemoryException in production, on a + /// payload comfortably UNDER every transport bound. The producer therefore chunks the WRITES. + /// But the prune is not chunkable: it asks "what is under this folder that the source no longer + /// carries", and a chunk that answered it from its own slice would delete the other chunks' + /// files. So the prune stays ONE authoritative pass and is handed the whole set — as PATHS, + /// which is all it ever needed and which costs nothing. + /// + /// Paths are collection-relative — the same {TargetPath}/{file} form + /// uses and the mirror compares against. + /// + public IReadOnlyList? MirrorKeepPaths { get; init; } } /// diff --git a/test/MeshWeaver.ContentCollections.Test/ContentSyncIsNeverBuiltWholeTest.cs b/test/MeshWeaver.ContentCollections.Test/ContentSyncIsNeverBuiltWholeTest.cs new file mode 100644 index 0000000000..773347ca3b --- /dev/null +++ b/test/MeshWeaver.ContentCollections.Test/ContentSyncIsNeverBuiltWholeTest.cs @@ -0,0 +1,285 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Reactive.Linq; +using System.Threading.Tasks; +using MeshWeaver.Fixture; +using MeshWeaver.Mesh; +using MeshWeaver.Messaging; +using Xunit; + +namespace MeshWeaver.ContentCollections.Test; + +/// +/// 🚨 THE PRODUCER MUST NOT BUILD ONE DELIVERY WHOLE — issue #2885 (and its twin #3046). +/// +/// The incident. A portal pod logged +/// OrleansRoutingService: Failed to deliver to import/xDAfkqsVUE-OMBHb0mVtSg with a +/// System.OutOfMemoryException raised at SharedArrayPool.Rent beneath +/// Utf8JsonWriter.TranscodeAndWriteRawValue, inside +/// MessageDeliveryConverter.Write. It recurred on 2026-09-02 against a second pod and a +/// second import hub, minutes after the AgenticBusiness/_GitSync node was written — i.e. +/// during a forced GitSync re-import. +/// +/// The arithmetic that identifies the producer. A GitSync import mirrors a Space's +/// git-committed content/** binaries with SyncContentFilesRequest, whose +/// Files carry the raw bytes INLINE. ContentAssetMapper.ToContentSyncs deliberately +/// emits ONE group per Space and SyncContentFilesBuilder.Post turned that group into ONE +/// message, so the delivery's size was the size of the Space's whole asset tree. +/// AgenticBusiness/content/ is 28,484,421 bytes of course video across 10 files: +/// base64 (×4/3) ≈ 38 MB of JSON, held as a UTF-16 RawJson.Content string ≈ 76 MB, which +/// TranscodeAndWriteRawValue then transcodes by renting up to 3 bytes per char ≈ 114 MB — +/// once per hop, and the report about a failed hop carries the body again. +/// AgenticEngineering/content/ is 106,070,300 bytes, whose base64 form exceeds Orleans' +/// 100 MiB MaxMessageBodySize outright: that Space's assets could not sync AT ALL. +/// +/// Why no bound is the fix. An OutOfMemoryException DURING serialization means +/// the ALLOCATION was the failure, so refusing the delivery afterwards is too late — the four +/// bounds already shipped for this family (#1890, #2897, #3018/#3032) all sit downstream of the +/// allocation, and the payload that killed the pod was in any case UNDER +/// MaxMessageBodySize. The only thing that removes the allocation is not making it: the +/// producer splits the write so that no single delivery ever carries the tree. +/// +/// What is asserted here. That a content sync larger than one delivery becomes +/// SEVERAL bounded deliveries whose union is exactly the supplied set, that exactly one of them +/// carries the authoritative mirror, and — the control — that a small sync is still ONE delivery +/// carrying no keep-set, byte-for-byte what it was before. +/// +public class ContentSyncIsNeverBuiltWholeTest(ITestOutputHelper output) : HubTestBase(output) +{ + /// + /// The tighter of the two transport ceilings the mesh declares + /// (DeliveryPayloadBounds.MemoryStreamBlockBytes), and the budget a content sync's + /// packaged bytes are chunked against. Restated here as a literal so the test fails if the + /// production constant is widened rather than silently following it. + /// + private const int PayloadBudgetBytes = 1 << 20; + + /// + /// 300,000 raw bytes → 400,000 base64 chars, so two files fit one budget and three do not. + /// Eight of them (2.4 MB raw, 3.2 MB packaged) therefore prove BATCHING, not merely + /// one-file-per-delivery. + /// + private const int FileBytes = 300_000; + + private const int FileCount = 8; + + private const string NodePath = "host/1"; + + /// Every the target hub actually received. + private readonly ConcurrentQueue received = new(); + + protected override MessageHubConfiguration ConfigureMesh(MessageHubConfiguration configuration) + => base.ConfigureMesh(configuration).WithTypes( + typeof(SyncContentFilesRequest), typeof(InlineContentFile), typeof(ImportContentResponse)); + + /// + /// Stands in for the Space-root hub: records what it was handed and answers, so the producer's + /// chunk sequence runs to completion and every delivery it built is observable. The real + /// handler's behaviour across a split is asserted by + /// . + /// + protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) + => base.ConfigureHost(configuration) + .WithTypes(typeof(SyncContentFilesRequest), typeof(InlineContentFile), typeof(ImportContentResponse)) + .WithHandler((hub, delivery) => + { + received.Enqueue(delivery.Message); + hub.Post(ImportContentResponse.Ok(delivery.Message.Files.Count), + o => o.ResponseFor(delivery)); + return delivery.Processed(); + }); + + protected override MessageHubConfiguration ConfigureClient(MessageHubConfiguration configuration) + => base.ConfigureClient(configuration).WithTypes( + typeof(SyncContentFilesRequest), typeof(InlineContentFile), typeof(ImportContentResponse)); + + /// The base64 length of a file's bytes — what the JSON payload actually costs. + private static long PackagedBytes(InlineContentFile file) + => 4L * ((file.Content.Length + 2) / 3) + file.Path.Length; + + private static long PackagedBytes(IEnumerable files) + => files.Sum(PackagedBytes); + + private static InlineContentFile[] Files(int count, int bytes) => + Enumerable.Range(0, count) + .Select(i => new InlineContentFile($"videos/asset-{i}.mp4", new byte[bytes])) + .ToArray(); + + /// + /// 🚨 THE FACT. A sync whose bytes exceed one delivery's budget is written as SEVERAL + /// deliveries, each within the budget plus at most one file, and their union is exactly the + /// supplied set. Before the fix this is ONE delivery of ~3.2 MB — the shape that OOM'd + /// production at 38 MB. + /// + [Fact] + public async Task A_sync_larger_than_one_delivery_is_split() + { + var files = Files(FileCount, FileBytes); + var whole = PackagedBytes(files); + whole.Should().BeGreaterThan(PayloadBudgetBytes, + "the fixture must be a set that CANNOT be carried by one delivery"); + + var response = await GetClient().SyncContentFiles(NodePath) + .To("content") + .Add(files) + .Mirror(true) + .Post() + .Timeout(TestTimeouts.Convergence) + .FirstAsync(); + + response.Success.Should().BeTrue(response.Error ?? string.Empty); + response.FilesImported.Should().Be(FileCount, "every file must still be written"); + + var deliveries = received.ToArray(); + deliveries.Length.Should().BeGreaterThan(1, + "a set of {0} packaged bytes cannot be one delivery — that is the allocation that " + + "threw OutOfMemoryException in production", whole); + + var largestFile = files.Max(PackagedBytes); + foreach (var delivery in deliveries) + PackagedBytes(delivery.Files).Should().BeLessThanOrEqualTo(PayloadBudgetBytes + largestFile, + "no delivery may exceed the budget by more than the one file that cannot be split"); + + deliveries.SelectMany(d => d.Files).Select(f => f.Path).ToArray() + .Should().Equal(files.Select(f => f.Path).ToArray(), + "the split must lose nothing, duplicate nothing and reorder nothing"); + } + + /// + /// 🚨 THE PRUNE IS STILL ONE AUTHORITATIVE PASS. A mirror deletes what the source no longer + /// carries, and a split must not turn that into several passes each measuring the collection + /// against its OWN chunk — which would make every chunk prune the others' files. Exactly one + /// delivery mirrors, and it names the FULL keep set rather than its own slice. + /// + [Fact] + public async Task Exactly_one_delivery_carries_the_mirror_and_it_names_the_whole_set() + { + var files = Files(FileCount, FileBytes); + + await GetClient().SyncContentFiles(NodePath) + .To("content", "TDD") + .Add(files) + .Mirror(true) + .SourceOwned(["TDD/videos/gone.mp4"]) + .Post() + .Timeout(TestTimeouts.Convergence) + .FirstAsync(); + + var deliveries = received.ToArray(); + deliveries.Length.Should().BeGreaterThan(1); + + var mirrors = deliveries.Where(d => d.Mirror).ToArray(); + mirrors.Should().ContainSingle("a split write still prunes exactly once"); + mirrors[0].SourceOwnedPaths.Should().Equal(["TDD/videos/gone.mp4"], + "the #435 preserve-user-uploads set belongs to the pruning pass"); + mirrors[0].MirrorKeepPaths.Should().Equal( + files.Select(f => $"TDD/{f.Path}").ToArray(), + "the prune must be measured against the WHOLE set, not the chunk it travels with"); + + deliveries.Where(d => !d.Mirror).Should().OnlyContain(d => d.SourceOwnedPaths == null, + "an additive chunk prunes nothing, so it carries no prune inputs"); + } + + /// + /// 🚨 THE CONTROL. A sync that fits stays exactly what it was: ONE delivery, mirroring + /// directly against its own Files, with no keep set on the wire. The split must be + /// invisible to every ordinary content sync. + /// + [Fact] + public async Task A_sync_that_fits_is_still_one_delivery() + { + var files = Files(3, 1_024); + PackagedBytes(files).Should().BeLessThan(PayloadBudgetBytes, "the control must fit"); + + await GetClient().SyncContentFiles(NodePath) + .To("content") + .Add(files) + .Mirror(true) + .Post() + .Timeout(TestTimeouts.Convergence) + .FirstAsync(); + + var deliveries = received.ToArray(); + deliveries.Should().ContainSingle("a sync that fits is not split"); + deliveries[0].Mirror.Should().BeTrue(); + deliveries[0].MirrorKeepPaths.Should().BeNull( + "an unsplit write is measured against its own Files, exactly as before"); + } +} + +/// +/// The split's semantic guard, driven through the REAL SyncContentFilesRequest handler and a +/// real file-system collection: after a write large enough to span several deliveries the folder +/// must mirror exactly the supplied set — every file written, the source's removed file pruned, and +/// a user upload the source never tracked PRESERVED (#435). +/// +/// This passes both before and after the fix, deliberately: it is what makes the split +/// safe to make, and it is the assertion that fails if the chunks are ever allowed to prune one +/// another. +/// +public class ContentSyncMirrorSurvivesTheSplitTest(ITestOutputHelper output) : HubTestBase(output) +{ + private const int FileBytes = 300_000; + private const int FileCount = 8; + private const string NodePath = "host/1"; + + private readonly string contentPath = Path.Combine( + AppContext.BaseDirectory, "Files", "SplitMirror", Guid.NewGuid().ToString("N")[..8]); + + protected override MessageHubConfiguration ConfigureMesh(MessageHubConfiguration configuration) + => base.ConfigureMesh(configuration).WithTypes( + typeof(SyncContentFilesRequest), typeof(InlineContentFile), typeof(ImportContentResponse)); + + protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) + { + Directory.CreateDirectory(contentPath); + return base.ConfigureHost(configuration) + .WithTypes(typeof(SyncContentFilesRequest), typeof(InlineContentFile), typeof(ImportContentResponse)) + .AddContentCollections() + .AddFileSystemContentCollection("content", _ => contentPath); + } + + protected override MessageHubConfiguration ConfigureClient(MessageHubConfiguration configuration) + => base.ConfigureClient(configuration).WithTypes( + typeof(SyncContentFilesRequest), typeof(InlineContentFile), typeof(ImportContentResponse)); + + [Fact] + public async Task A_split_mirror_writes_everything_prunes_the_stale_and_keeps_the_upload() + { + Directory.CreateDirectory(Path.Combine(contentPath, "videos")); + // What the SOURCE carried last time and no longer carries — must be pruned. + await File.WriteAllBytesAsync(Path.Combine(contentPath, "videos", "retired.mp4"), new byte[16]); + // What a USER uploaded and the source never tracked — must survive (#435). + await File.WriteAllBytesAsync(Path.Combine(contentPath, "videos", "user-upload.mp4"), new byte[16]); + + var files = Enumerable.Range(0, FileCount) + .Select(i => new InlineContentFile($"videos/asset-{i}.mp4", new byte[FileBytes])) + .ToArray(); + + var response = await GetClient().SyncContentFiles(NodePath) + .To("content") + .Add(files) + .Mirror(true) + .SourceOwned(["videos/retired.mp4"]) + .Post() + .Timeout(TestTimeouts.Convergence) + .FirstAsync(); + + response.Success.Should().BeTrue(response.Error ?? string.Empty); + response.FilesImported.Should().Be(FileCount); + + var onDisk = Directory.GetFiles(Path.Combine(contentPath, "videos")) + .Select(Path.GetFileName) + .ToArray(); + + onDisk.Should().Contain(files.Select(f => Path.GetFileName(f.Path)), + "every supplied file must land, whichever delivery carried it"); + onDisk.Should().NotContain("retired.mp4", + "a source-owned file the source no longer carries is pruned — once, by the mirror pass"); + onDisk.Should().Contain("user-upload.mp4", + "a file the source never owned survives a mirror (#435) — and survives the split"); + } +}