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)