From 75c145c2bf1f4bec5a46bd34727b70f59827d63d Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 3 Aug 2026 08:51:54 -0500 Subject: [PATCH] Skip the body copy on a StreamOne 304, and close the revision-ETag test gaps Follow-up to #5121. The ETag values and 304 semantics are unchanged. Efficiency: - Read mt_version off the row BEFORE the payload and consult a shouldWriteBody predicate, so a conditional-request hit never grows the pooled MemoryStream or copies the document. Column order is free here: readers access by GetOrdinal and Marten never opens them with CommandBehavior.SequentialAccess. The row still comes back from Postgres, so this saves the copy, not the read. Applies to the Guid path identically. - Drop the Convert.ToInt64(GetFieldValueAsync()) box on the revision read in favor of a GetFieldType branch, keeping the #4614 both-widths handling. - Collapse StreamOneWithVersion/StreamOneWithRevision, which differed only in the CLR type read back, into one method behind a numericRevision flag. Tests: - Pin the single round trip on the revision path (the Guid path already had one). - Pin that a 304 leaves the response buffer empty. - Cover ILongVersioned (bigint) alongside IRevisioned (integer) on purpose. - Cover 404-emits-no-ETag and EmitETag = false on a revisioned document. - Cover the IRevisioned + UseOptimisticConcurrency route into #5121's guard, which is the likeliest real-world one and was untested. Verified against master that it previously failed as DDL with two mt_version columns. Docs: - Correct the claim that EventProjection outputs emit no ETag. They are not aggregate projection targets, so ProjectionDocumentPolicy leaves them on the plain-document default and they emit a Guid ETag that changes on every projection write -- a usable cache validator, but not a stream version. - Document #5121's fail-fast guard, which shipped undocumented. Co-Authored-By: Claude Opus 5 --- docs/documents/aspnetcore.md | 13 +- docs/documents/concurrency.md | 18 ++ ...urrency_on_projection_target_fails_fast.cs | 24 +++ src/IssueService/Startup.cs | 8 + src/IssueService/StreamingMinimalEndpoints.cs | 62 +++++++ .../streaming_result_types_tests.cs | 165 ++++++++++++++++++ src/Marten.AspNetCore/QueryableExtensions.cs | 49 ++++-- .../Sessions/QuerySession.Execution.cs | 22 +-- src/Marten/Linq/MartenLinqQueryProvider.cs | 44 ++--- src/Marten/Linq/MartenLinqQueryable.cs | 9 +- .../Services/JsonStreamingExtensions.cs | 80 +++++---- 11 files changed, 397 insertions(+), 97 deletions(-) diff --git a/docs/documents/aspnetcore.md b/docs/documents/aspnetcore.md index c3fccc0853..141a04e4d7 100644 --- a/docs/documents/aspnetcore.md +++ b/docs/documents/aspnetcore.md @@ -466,15 +466,20 @@ still current: document), formatted as a quoted GUID, e.g. `"3f2504e0-4f89-11d3-9a0c-0305e82c3301"`. The `mt_version` value is read **inline with the document in the same single database round trip** (piggy-backed onto the streaming query), so enabling the - ETag adds no extra query. Because the document is streamed in that one round - trip, a `304` on `StreamOne` saves response bandwidth but not the read. + ETag adds no extra query. The version is read off the row _before_ the document + payload, so a `304` skips buffering a body it would only discard — but the row + itself still comes back from Postgres, so a `304` on `StreamOne` saves the + response and the copy, not the read. - For documents using [numeric revisioning](/documents/concurrency) instead of the Guid version — `IRevisioned`/`ILongVersioned` types, and every aggregate-projection target (snapshots, single- and multi-stream projections), since Marten forces numeric revisions on those — the `StreamOne` ETag is the numeric `mt_version`, formatted as a quoted integer, e.g. `"3"`. Documents written by an - `EventProjection` are not forced into numeric revisions and emit no ETag unless - they opt into a versioning flavor themselves. For a `SingleStreamProjection` + `EventProjection` are **not** aggregate-projection targets, so Marten does not + force numeric revisions on them; they keep the plain-document default and emit a + quoted GUID ETag that changes on every projection write. That is a usable cache + validator, but it is not a stream version and does not line up with what + `StreamAggregate` serves. For a `SingleStreamProjection` target the revision is the source stream's version, so serving the read model through `StreamOne` produces the same ETag that `StreamAggregate` would serve for the stream itself; for an Inline-lifecycle projection, clients can diff --git a/docs/documents/concurrency.md b/docs/documents/concurrency.md index 4dc8eb5ee4..5e0f972432 100644 --- a/docs/documents/concurrency.md +++ b/docs/documents/concurrency.md @@ -193,6 +193,24 @@ same session. Prefer using `UpdateRevision()` if you try to continuously update `IRevisioned.Version` is an `int` — the right choice for an ordinary per-document revision counter. For documents projected from a `MultiStreamProjection` whose `Version` is the global **event sequence number**, the value can exceed `Int32.MaxValue`; implement `ILongVersioned` (with a `long Version`) instead. Both opt the document into numeric revisioning; the only difference is the column type and member width: `IRevisioned` stores its version in an `integer` (`mt_version`) column, while `ILongVersioned` uses a `bigint` column. A `MultiStreamProjection`-derived document that implements `IRevisioned` (int) will overflow on the `bigint → int` read once its version exceeds `Int32` — use `ILongVersioned` for those. ::: +::: warning A document cannot use both flavors +Guid optimistic concurrency and numeric revisioning both store their value in the same physical +`mt_version` column, so a document type can only have one of them. Marten throws an +`InvalidDocumentException` at bootstrap if a mapping ends up with both enabled, rather than letting +it reach the database as DDL with two `mt_version` columns. + +The two ways to trip this are worth knowing, because neither reads as "I asked for both": + +- Calling `UseOptimisticConcurrency(true)` on a type that already has numeric revisions — because it + implements `IRevisioned`/`ILongVersioned`, carries a `[Version]` member, or is an + aggregate-projection target (Marten forces numeric revisions on those). Fluent configuration runs + after those policies and layers on top of them rather than replacing them. +- Calling `UseNumericRevisions(true)` and then `UseOptimisticConcurrency(true)` on the same type. + +Remove the call for the flavor you do not want. Aggregate-projection targets always use numeric +revisions and cannot opt into `UseOptimisticConcurrency`. +::: + or finally by adding the `[Version]` attribute to a public member on the document type to opt into the `UseNumericRevisions` behavior on the parent type with the decorated member being tracked as the version number as shown in this sample: diff --git a/src/EventSourcingTests/Bugs/optimistic_concurrency_on_projection_target_fails_fast.cs b/src/EventSourcingTests/Bugs/optimistic_concurrency_on_projection_target_fails_fast.cs index a2b3a7b113..bd4450c28e 100644 --- a/src/EventSourcingTests/Bugs/optimistic_concurrency_on_projection_target_fails_fast.cs +++ b/src/EventSourcingTests/Bugs/optimistic_concurrency_on_projection_target_fails_fast.cs @@ -1,6 +1,8 @@ +using System; using EventSourcingTests.Aggregation; using JasperFx.Events.Projections; using Marten.Exceptions; +using Marten.Metadata; using Marten.Testing.Documents; using Marten.Testing.Harness; using Shouldly; @@ -49,4 +51,26 @@ public void switching_a_plain_document_from_numeric_revisions_to_optimistic_conc Should.Throw( () => theStore.Options.Storage.MappingFor(typeof(Target))); } + + [Fact] + public void interface_driven_revisions_plus_optimistic_concurrency_fails_fast() + { + // The likeliest real-world route into the invalid state, and the one neither test above + // covers: nothing in the configuration says "numeric revisions" — VersionedPolicy turns + // them on because the document implements IRevisioned, and the fluent call then re-enables + // the Guid version on top. Before the guard this reached the database as DDL with two + // mt_version columns. + StoreOptions(opts => opts.Schema.For().UseOptimisticConcurrency(true)); + + var ex = Should.Throw( + () => theStore.Options.Storage.MappingFor(typeof(RevisionedDoc))); + + ex.Message.ShouldContain(nameof(RevisionedDoc)); + } + + public class RevisionedDoc: IRevisioned + { + public Guid Id { get; set; } + public int Version { get; set; } + } } diff --git a/src/IssueService/Startup.cs b/src/IssueService/Startup.cs index a6c0c65fe8..63c93da836 100644 --- a/src/IssueService/Startup.cs +++ b/src/IssueService/Startup.cs @@ -4,6 +4,7 @@ using System.Threading.Tasks; using IssueService.Controllers; using JasperFx.Events; +using JasperFx.Events.Projections; using Marten; using Marten.Events.Projections; using Marten.Testing.Harness; @@ -56,7 +57,14 @@ public void ConfigureServices(IServiceCollection services) } else { + // NOTE: Order is only a projection target under Guid stream identity, so the + // /minimal/order-doc endpoint only carries a revision-derived ETag on hosts built + // from this branch. Tests asserting that ETag must use the Guid-identity fixture. options.Projections.Snapshot(SnapshotLifecycle.Inline); + + // An EventProjection (not an aggregate projection) so its output document is NOT + // swept into numeric revisions by ProjectionDocumentPolicy. + options.Projections.Add(ProjectionLifecycle.Inline); } return options; diff --git a/src/IssueService/StreamingMinimalEndpoints.cs b/src/IssueService/StreamingMinimalEndpoints.cs index 996b3adfee..4a409fa30b 100644 --- a/src/IssueService/StreamingMinimalEndpoints.cs +++ b/src/IssueService/StreamingMinimalEndpoints.cs @@ -1,8 +1,11 @@ using System; using System.Linq; using IssueService.Controllers; +using JasperFx.Events; using Marten; using Marten.AspNetCore; +using Marten.Events.Projections; +using Marten.Metadata; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Routing; @@ -68,6 +71,30 @@ public static IEndpointRouteBuilder MapStreamingMinimalEndpoints(this IEndpointR (Guid id, IQuerySession session) => new StreamOne(session.Query().Where(x => x.Id == id))); + // EmitETag = false opt-out on a numeric-revision document — proves the opt-out short-circuits + // before the revision flavor is ever consulted, not just for the Guid flavor. + app.MapGet("/minimal/revisioned/{id:guid}/no-etag", + (Guid id, IQuerySession session) + => new StreamOne(session.Query().Where(x => x.Id == id)) + { + EmitETag = false + }); + + // Plain document using the 64-bit revision flavor via ILongVersioned — the shape a + // MultiStreamProjection target takes, where the revision is a per-document counter rather + // than a stream version, and the mt_version column stays bigint. + app.MapGet("/minimal/long-versioned/{id:guid}", + (Guid id, IQuerySession session) + => new StreamOne( + session.Query().Where(x => x.Id == id))); + + // Document written by an EventProjection. ProjectionDocumentPolicy only forces numeric + // revisions onto *aggregate* projection targets, so this one keeps the default Guid + // version metadata — see the Alba test for what that means for its ETag. + app.MapGet("/minimal/event-projection-doc/{id:guid}", + (Guid id, IQuerySession session) + => new StreamOne(session.Query().Where(x => x.Id == id))); + // --- StreamMany --- app.MapGet("/minimal/issues/open", @@ -206,3 +233,38 @@ public class RevisionedIssueNote: IRevisioned public string Name { get; set; } public int Version { get; set; } } + +/// +/// A plain (non-projection) document using the 64-bit revision flavor via +/// — the shape a MultiStreamProjection target takes. Its +/// mt_version column stays bigint (unlike , which +/// #4614 narrows to integer), so the pair covers both widths the revision read must handle. +/// +public class LongVersionedIssueNote: ILongVersioned +{ + public Guid Id { get; set; } + public string Name { get; set; } + public long Version { get; set; } +} + +/// +/// Output document of , an EventProjection rather than an +/// aggregate projection. ProjectionDocumentPolicy only forces numeric revisions onto +/// aggregate targets, so this type is left with whatever versioning a plain document gets. +/// +public class OrderTouch +{ + public Guid Id { get; set; } + public string Description { get; set; } +} + +/// +/// An EventProjection (not an aggregate projection) writing +/// documents keyed by stream id. Declared partial so the JasperFx.Events source generator +/// can emit its dispatcher for the conventional Create method. +/// +public partial class OrderTouchProjection: EventProjection +{ + public OrderTouch Create(IEvent e) + => new() { Id = e.StreamId, Description = e.Data.Description }; +} diff --git a/src/Marten.AspNetCore.Testing/streaming_result_types_tests.cs b/src/Marten.AspNetCore.Testing/streaming_result_types_tests.cs index 3c13375d5f..e15fb6f5cb 100644 --- a/src/Marten.AspNetCore.Testing/streaming_result_types_tests.cs +++ b/src/Marten.AspNetCore.Testing/streaming_result_types_tests.cs @@ -421,6 +421,171 @@ await theHost.Scenario(s => }); } + [Fact] + public async Task stream_one_emits_revision_etag_for_long_versioned_document() + { + // ILongVersioned keeps the default bigint mt_version column, where IRevisioned narrows it + // to integer (#4614). Pinning both proves the revision read copes with either width, and + // that a multi-stream-shaped target emits its own per-document counter as a valid ETag. + var note = new LongVersionedIssueNote { Id = Guid.NewGuid(), Name = "long rev" }; + await using (var session = theHost.Services.GetRequiredService().LightweightSession()) + { + session.Store(note); + await session.SaveChangesAsync(); + } + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/long-versioned/{note.Id}"); + s.StatusCodeShouldBe(200); + s.ContentTypeShouldBe("application/json"); + }); + + result.Context.Response.Headers["ETag"].ToString().ShouldBe("\"1\""); + + await theHost.Scenario(s => + { + s.Get.Url($"/minimal/long-versioned/{note.Id}"); + s.WithRequestHeader("If-None-Match", "\"1\""); + s.StatusCodeShouldBe(304); + }); + } + + [Fact] + public async Task stream_one_returns_404_without_etag_for_a_revisioned_document() + { + // The 404 branch runs before any ETag is formatted, but that was only pinned on the Guid + // path — a projection-target miss must not leak a header either. + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order-doc/{Guid.NewGuid()}"); + s.StatusCodeShouldBe(404); + }); + + result.Context.Response.Headers.ContainsKey("ETag").ShouldBeFalse(); + } + + [Fact] + public async Task stream_one_suppresses_etag_on_a_revisioned_document_when_emit_etag_is_false() + { + var note = new RevisionedIssueNote { Id = Guid.NewGuid(), Name = "opted out" }; + await using (var session = theHost.Services.GetRequiredService().LightweightSession()) + { + session.Store(note); + await session.SaveChangesAsync(); + } + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/revisioned/{note.Id}/no-etag"); + s.StatusCodeShouldBe(200); + }); + + result.Context.Response.Headers.ContainsKey("ETag").ShouldBeFalse(); + + // The opt-out must still serve the document, not just drop the header. + result.ReadAsJson().Name.ShouldBe("opted out"); + } + + [Fact] + public async Task stream_one_emits_a_guid_etag_for_an_event_projection_output_document() + { + // ProjectionDocumentPolicy only forces numeric revisions onto *aggregate* projection + // targets. An EventProjection's output keeps the plain-document default, which is Guid + // version metadata — so it does emit an ETag, just not a stream-derived one. That ETag + // is opaque and changes on every projection write, so it is only safe as a cache + // validator, never as a stream-version equivalent. + var orderId = Guid.NewGuid(); + await using (var session = theHost.Services.GetRequiredService().LightweightSession()) + { + session.Events.StartStream(orderId, new OrderPlaced("touched", 5.00m)); + await session.SaveChangesAsync(); + } + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/event-projection-doc/{orderId}"); + s.StatusCodeShouldBe(200); + }); + + var etag = result.Context.Response.Headers["ETag"].ToString(); + etag.ShouldNotBeNullOrEmpty(); + + // A Guid ETag, not the stream version the aggregate target would have served. + Guid.TryParse(etag.Trim('"'), out _).ShouldBeTrue(); + etag.ShouldNotBe("\"1\""); + + await theHost.Scenario(s => + { + s.Get.Url($"/minimal/event-projection-doc/{orderId}"); + s.WithRequestHeader("If-None-Match", etag); + s.StatusCodeShouldBe(304); + }); + } + + [Fact] + public async Task stream_one_with_revision_etag_executes_a_single_db_command() + { + // Companion to stream_one_with_etag_executes_a_single_db_command, which only covered the + // Guid flavor. The revision flavor is the projection read-model path — the hot one — so + // pin that it also resolves the document AND its ETag in ONE round trip. + var store = theHost.Services.GetRequiredService(); + + var orderId = Guid.NewGuid(); + await using (var session = store.LightweightSession()) + { + session.Events.StartStream(orderId, new OrderPlaced("single command", 1.00m)); + await session.SaveChangesAsync(); + } + + await using var query = store.QuerySession(); + var logger = new CommandCountingLogger(); + query.Logger = logger; + + // Warm up storage-existence checks on this session so they don't count against us. + await query.Query().Where(x => x.Id == Guid.NewGuid()) + .StreamJsonFirstOrDefault(new MemoryStream()); + logger.Count = 0; + + var context = new DefaultHttpContext { Response = { Body = new MemoryStream() } }; + + await query.Query().Where(x => x.Id == orderId) + .WriteSingle(context, emitETag: true); + + context.Response.StatusCode.ShouldBe(200); + context.Response.Headers["ETag"].ToString().ShouldBe("\"1\""); + logger.Count.ShouldBe(1); + } + + [Fact] + public async Task stream_one_does_not_buffer_the_document_body_on_a_304() + { + // A conditional-request hit reads the version off the row and then declines the payload, + // so nothing is copied into the response buffer. The read itself still happens — the row + // comes back either way — but the document copy does not. + var store = theHost.Services.GetRequiredService(); + + var orderId = Guid.NewGuid(); + await using (var session = store.LightweightSession()) + { + session.Events.StartStream(orderId, new OrderPlaced(new string('x', 20_000), 1.00m)); + await session.SaveChangesAsync(); + } + + await using var query = store.QuerySession(); + + var body = new MemoryStream(); + var context = new DefaultHttpContext { Response = { Body = body } }; + context.Request.Headers["If-None-Match"] = "\"1\""; + + await query.Query().Where(x => x.Id == orderId).WriteSingle(context, emitETag: true); + + context.Response.StatusCode.ShouldBe(StatusCodes.Status304NotModified); + context.Response.Headers["ETag"].ToString().ShouldBe("\"1\""); + context.Response.ContentLength.ShouldBe(0); + body.Length.ShouldBe(0); + } + // ───────────────────────── StreamMany ───────────────────────── [Fact] diff --git a/src/Marten.AspNetCore/QueryableExtensions.cs b/src/Marten.AspNetCore/QueryableExtensions.cs index b5693ad7fa..8185705208 100644 --- a/src/Marten.AspNetCore/QueryableExtensions.cs +++ b/src/Marten.AspNetCore/QueryableExtensions.cs @@ -23,8 +23,10 @@ public static class QueryableExtensions /// integer for numeric-revision documents (projection targets, IRevisioned types), where /// a SingleStreamProjection target's revision is the source stream's version. If the /// incoming request's If-None-Match header matches that value, a 304 Not Modified - /// is written instead, with an empty body. Document types with neither version nor revision - /// metadata enabled (no mt_version column) emit no ETag. + /// is written instead, with an empty body — and because the version is read off the row before + /// the payload, the document is never copied into the response buffer on that path. Document + /// types with neither version nor revision metadata enabled (no mt_version column) emit + /// no ETag. /// /// /// @@ -60,8 +62,12 @@ public static async Task WriteSingle( // Fetch the document JSON and its mt_version in a single round trip. The version rides // on the same streaming query (see MartenLinqQueryProvider.StreamOneWithVersion), so there // is no follow-up MetadataForAsync query and no re-deserialization of the buffered JSON. + // The predicate runs once the version is known and before the payload is copied, so a + // cache hit skips buffering a body that would only be discarded. var result = await ((MartenLinqQueryable)queryable) - .StreamJsonFirstOrDefaultWithVersion(stream, context.RequestAborted) + .StreamJsonFirstOrDefaultWithVersion(stream, + (version, revision) => !matchesIfNoneMatch(context, formatETag(version, revision)), + context.RequestAborted) .ConfigureAwait(false); if (!result.Found) @@ -71,28 +77,37 @@ public static async Task WriteSingle( return; } - var etag = result.Version.HasValue - ? ETagHelpers.Format(result.Version.Value) - : result.Revision.HasValue - ? ETagHelpers.Format(result.Revision.Value) - : null; - + var etag = formatETag(result.Version, result.Revision); if (etag != null) { - if (ETagHelpers.IfNoneMatchMatches(context, etag)) - { - context.Response.StatusCode = StatusCodes.Status304NotModified; - context.Response.Headers["ETag"] = etag; - context.Response.ContentLength = 0; - return; - } - context.Response.Headers["ETag"] = etag; } + if (!result.BodyWritten) + { + context.Response.StatusCode = StatusCodes.Status304NotModified; + context.Response.ContentLength = 0; + return; + } + await writeBufferedBody(context, stream, contentType, onFoundStatus).ConfigureAwait(false); } + /// + /// Format whichever mt_version flavor came back as a quoted strong ETag, or null when the + /// document type carries neither (no mt_version column) — in which case no ETag is emitted + /// and no conditional request can match. + /// + private static string? formatETag(Guid? version, long? revision) + => version.HasValue + ? ETagHelpers.Format(version.Value) + : revision.HasValue + ? ETagHelpers.Format(revision.Value) + : null; + + private static bool matchesIfNoneMatch(HttpContext context, string? etag) + => etag != null && ETagHelpers.IfNoneMatchMatches(context, etag); + private static async Task writeBufferedBody(HttpContext context, System.IO.Stream stream, string contentType, int onFoundStatus) { diff --git a/src/Marten/Internal/Sessions/QuerySession.Execution.cs b/src/Marten/Internal/Sessions/QuerySession.Execution.cs index b2a3daed36..da30e79079 100644 --- a/src/Marten/Internal/Sessions/QuerySession.Execution.cs +++ b/src/Marten/Internal/Sessions/QuerySession.Execution.cs @@ -87,29 +87,15 @@ internal async Task StreamOne(DbCommand command, Stream stream, Cancellati } } - internal async Task<(bool found, Guid? version)> StreamOneWithVersion(DbCommand command, Stream stream, - CancellationToken token) + internal async Task StreamOneWithVersion(DbCommand command, Stream stream, + bool numericRevision, Func? shouldWriteBody, CancellationToken token) { await using var reader = await ExecuteReaderAsync(command, token).ConfigureAwait(false); try { - return await reader.StreamOneWithVersion(stream, token).ConfigureAwait(false); - } - finally - { - await reader.CloseAsync().ConfigureAwait(false); - } - } - - internal async Task<(bool found, long? revision)> StreamOneWithRevision(DbCommand command, Stream stream, - CancellationToken token) - { - await using var reader = await ExecuteReaderAsync(command, token).ConfigureAwait(false); - - try - { - return await reader.StreamOneWithRevision(stream, token).ConfigureAwait(false); + return await reader.StreamOneWithVersion(stream, numericRevision, shouldWriteBody, token) + .ConfigureAwait(false); } finally { diff --git a/src/Marten/Linq/MartenLinqQueryProvider.cs b/src/Marten/Linq/MartenLinqQueryProvider.cs index 01461d97b6..82476d5d85 100644 --- a/src/Marten/Linq/MartenLinqQueryProvider.cs +++ b/src/Marten/Linq/MartenLinqQueryProvider.cs @@ -31,8 +31,13 @@ internal record WaitForAggregate(TimeSpan Timeout, NonStaleDataTimeoutMode Timeo /// types) carries a value; both are null when the document type has no mt_version column /// (neither metadata flavor enabled) or the value was SQL NULL — in which case no ETag /// should be emitted. +/// +/// is false when the caller's shouldWriteBody predicate declined +/// the payload after seeing the version — the conditional-request (304) case, where nothing +/// was copied into the destination stream. +/// /// -internal readonly record struct StreamOneJsonResult(bool Found, Guid? Version, long? Revision); +internal readonly record struct StreamOneJsonResult(bool Found, Guid? Version, long? Revision, bool BodyWritten); internal class MartenLinqQueryProvider: IQueryProvider { @@ -281,9 +286,15 @@ public async Task StreamOne(Expression expression, Stream destination, Can /// When the document type has no mt_version column (neither /// metadata flavor enabled), the column is not appended and the result carries neither a /// version nor a revision so the caller emits no ETag. + /// + /// is consulted after the version is read but before the + /// document payload is copied into , so a conditional-request + /// caller answering 304 pays for neither the copy nor the buffer growth. It is not + /// consulted on the no-version path, where there is nothing to decide on. + /// /// public async Task StreamOneWithVersion(Expression expression, Stream destination, - CancellationToken token) where T : notnull + Func? shouldWriteBody, CancellationToken token) where T : notnull { var parser = new LinqQueryParser(this, _session, expression); var statements = parser.BuildStatements(); @@ -296,35 +307,26 @@ public async Task StreamOneWithVersion(Expression expres var mapping = _session.Options.Storage.FindMapping(typeof(T)) as DocumentMapping; - if (mapping is { Metadata.Version.Enabled: true }) - { - main.SelectClause = new VersionSelectClause(main.SelectClause); - - var command = statement.BuildCommand(_session); - var (streamed, version) = await _session.StreamOneWithVersion(command, destination, token) - .ConfigureAwait(false); - - return new StreamOneJsonResult(streamed, version, null); - } + // Both flavors keep their value in the same physical mt_version column, so one + // piggy-backed select serves them; only the CLR type read back differs. For a + // SingleStreamProjection target the revision *is* the source stream's version, making the + // resulting ETag byte-for-byte the one StreamAggregate serves for the same stream. + var numericRevision = mapping is { Metadata.Revision.Enabled: true }; - // Numeric-revision documents keep their revision in the same physical mt_version column, - // so the identical piggy-backed select serves them — the value just reads back as a - // number rather than a uuid. For a SingleStreamProjection target the revision *is* the - // source stream's version, making the resulting ETag byte-for-byte the one StreamAggregate - // serves for the same stream. - if (mapping is { Metadata.Revision.Enabled: true }) + if (numericRevision || mapping is { Metadata.Version.Enabled: true }) { main.SelectClause = new VersionSelectClause(main.SelectClause); var command = statement.BuildCommand(_session); - var (streamed, revision) = await _session.StreamOneWithRevision(command, destination, token) + var result = await _session + .StreamOneWithVersion(command, destination, numericRevision, shouldWriteBody, token) .ConfigureAwait(false); - return new StreamOneJsonResult(streamed, null, revision); + return new StreamOneJsonResult(result.Found, result.Version, result.Revision, result.BodyWritten); } var plainCommand = statement.BuildCommand(_session); var found = await _session.StreamOne(plainCommand, destination, token).ConfigureAwait(false); - return new StreamOneJsonResult(found, null, null); + return new StreamOneJsonResult(found, null, null, found); } } diff --git a/src/Marten/Linq/MartenLinqQueryable.cs b/src/Marten/Linq/MartenLinqQueryable.cs index 14f3b0d139..acf6240b3b 100644 --- a/src/Marten/Linq/MartenLinqQueryable.cs +++ b/src/Marten/Linq/MartenLinqQueryable.cs @@ -311,10 +311,15 @@ public Task StreamJsonFirstOrDefault(Stream destination, CancellationToken /// mt_version in the SAME database round trip (see /// ). Used by the ASP.NET Core /// StreamOne ETag support to avoid a follow-up metadata query. + /// + /// sees the version before the payload is copied, so a + /// caller that answers 304 Not Modified can decline the body it would only discard. + /// /// - internal Task StreamJsonFirstOrDefaultWithVersion(Stream destination, CancellationToken token) + internal Task StreamJsonFirstOrDefaultWithVersion(Stream destination, + Func? shouldWriteBody, CancellationToken token) { - return MartenProvider.StreamOneWithVersion(Expression, destination, token); + return MartenProvider.StreamOneWithVersion(Expression, destination, shouldWriteBody, token); } public Task StreamJsonSingle(Stream destination, CancellationToken token) diff --git a/src/Marten/Services/JsonStreamingExtensions.cs b/src/Marten/Services/JsonStreamingExtensions.cs index d347b1f0c0..69a62de98c 100644 --- a/src/Marten/Services/JsonStreamingExtensions.cs +++ b/src/Marten/Services/JsonStreamingExtensions.cs @@ -1,6 +1,5 @@ #nullable enable using System; -using System.Globalization; using System.IO; using System.Text; using System.Text.Json; @@ -13,6 +12,14 @@ namespace Marten.Services; +/// +/// Outcome of a single-row read that pairs a document's raw JSON with the piggy-backed +/// mt_version value. is false when the caller's +/// shouldWriteBody predicate declined the payload after seeing the version — +/// the conditional-request (304) case, where copying the document would be wasted work. +/// +internal readonly record struct StreamOneReadResult(bool Found, Guid? Version, long? Revision, bool BodyWritten); + internal static class JsonStreamingExtensions { internal static readonly byte[] LeftBracket = Encoding.Default.GetBytes("["); @@ -37,56 +44,59 @@ internal static async Task StreamOne(this DbDataReader reader, Stream strea /// /// Streams the first row's data column to (as /// does) AND reads the piggy-backed mt_version value - /// aliased as , + /// aliased as , /// so a single-document JSON stream and its version come back in one round trip. - /// Returns found = false when the query matched no row, and a null - /// version when the version column value was SQL NULL. + /// + /// The same physical column carries either flavor: a Guid under optimistic concurrency, or a + /// number under revisioning (projection targets, IRevisioned/ILongVersioned types). + /// picks which one to materialize; the numeric column is + /// bigint by default but integer for IRevisioned-backed documents (#4614), + /// so the accessor matching the reported width is used rather than boxing through object. + /// + /// + /// The version is read BEFORE the payload. Marten never opens readers with + /// CommandBehavior.SequentialAccess, so the row is fully buffered and column order does + /// not constrain read order — which lets veto the document + /// copy once the version is known. Returns Found = false when the query matched no row, + /// and null for both flavors when the column value was SQL NULL. + /// /// - internal static async Task<(bool found, Guid? version)> StreamOneWithVersion(this DbDataReader reader, - Stream stream, CancellationToken token) + internal static async Task StreamOneWithVersion(this DbDataReader reader, + Stream stream, bool numericRevision, Func? shouldWriteBody, CancellationToken token) { if (!await reader.ReadAsync(token).ConfigureAwait(false)) { - return (false, null); + return new StreamOneReadResult(false, null, null, false); } - var dataOrdinal = reader.GetOrdinal("data"); - await reader.WriteJsonValueAsync(dataOrdinal, stream, token).ConfigureAwait(false); - var versionOrdinal = reader.GetOrdinal(Marten.Linq.SqlGeneration.VersionSelectClause.VersionAlias); - Guid? version = await reader.IsDBNullAsync(versionOrdinal, token).ConfigureAwait(false) - ? null - : await reader.GetFieldValueAsync(versionOrdinal, token).ConfigureAwait(false); - return (true, version); - } + Guid? version = null; + long? revision = null; - /// - /// Numeric-revision counterpart of : streams the first - /// row's data column and reads the piggy-backed numeric mt_version value. - /// The column is bigint by default but integer for IRevisioned-backed - /// documents (#4614), so the value is materialized at whatever width came back and widened - /// to long rather than read through a single generic accessor. - /// - internal static async Task<(bool found, long? revision)> StreamOneWithRevision(this DbDataReader reader, - Stream stream, CancellationToken token) - { - if (!await reader.ReadAsync(token).ConfigureAwait(false)) + if (!await reader.IsDBNullAsync(versionOrdinal, token).ConfigureAwait(false)) { - return (false, null); + if (numericRevision) + { + revision = reader.GetFieldType(versionOrdinal) == typeof(int) + ? await reader.GetFieldValueAsync(versionOrdinal, token).ConfigureAwait(false) + : await reader.GetFieldValueAsync(versionOrdinal, token).ConfigureAwait(false); + } + else + { + version = await reader.GetFieldValueAsync(versionOrdinal, token).ConfigureAwait(false); + } + } + + if (shouldWriteBody != null && !shouldWriteBody(version, revision)) + { + return new StreamOneReadResult(true, version, revision, false); } var dataOrdinal = reader.GetOrdinal("data"); await reader.WriteJsonValueAsync(dataOrdinal, stream, token).ConfigureAwait(false); - var revisionOrdinal = reader.GetOrdinal(Marten.Linq.SqlGeneration.VersionSelectClause.VersionAlias); - long? revision = await reader.IsDBNullAsync(revisionOrdinal, token).ConfigureAwait(false) - ? null - : Convert.ToInt64( - await reader.GetFieldValueAsync(revisionOrdinal, token).ConfigureAwait(false), - CultureInfo.InvariantCulture); - - return (true, revision); + return new StreamOneReadResult(true, version, revision, true); } internal static ValueTask WriteBytes(this Stream stream, byte[] bytes, CancellationToken token)