Derive the StreamOne ETag from the numeric revision for revisioned documents

docs/documents/aspnetcore.md

@@ -466,9 +466,29 @@ 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. Document types whose version metadata is disabled (no
-  `mt_version` column) simply emit no ETag. Because the document is streamed in that
-  one round trip, a `304` on `StreamOne<T>` saves response bandwidth but not the read.
+  ETag adds no extra query. Because the document is streamed in that one round
+  trip, a `304` on `StreamOne<T>` saves response bandwidth but 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<T>` 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`
+  target the revision is the source stream's version, so serving the read model
+  through `StreamOne<T>` produces the same ETag that `StreamAggregate<T>` would
+  serve for the stream itself; for an Inline-lifecycle projection, clients can
+  switch between the two read styles without invalidating their caches (an
+  Async-lifecycle document can lag the stream head, so the two styles may briefly
+  disagree). Multi-stream and `ILongVersioned` targets emit their own monotonic
+  per-document revision, which is a valid ETag but not a stream version. Note that
+  revision-derived ETags are deterministic: a projection-logic change plus rebuild
+  that does not advance the stream does not invalidate previously cached responses
+  (bumping `ProjectionVersion` does not reset the revision either), matching the
+  `StreamAggregate<T>` semantics. The compiled-query overload
+  (`StreamOne<TDoc, TOut>`) does not participate in ETag/`304` handling. Document
+  types with neither version flavor enabled (no `mt_version` column) simply emit
+  no ETag.
 - For `StreamAggregate<T>`, the ETag is derived from the event stream's
   version (a `long`), formatted as a quoted integer, e.g. `"42"`. The version
   is looked up before the aggregate is folded, so a cache hit (`304`) skips

src/EventSourcingTests/Bugs/optimistic_concurrency_on_projection_target_fails_fast.cs

@@ -0,0 +1,52 @@
+using EventSourcingTests.Aggregation;
+using JasperFx.Events.Projections;
+using Marten.Exceptions;
+using Marten.Testing.Documents;
+using Marten.Testing.Harness;
+using Shouldly;
+using Xunit;
+
+namespace EventSourcingTests.Bugs;
+
+/// <summary>
+/// Companion guard to Bug_2978: the Guid version column and the numeric revision column are
+/// two flavors of the same physical <c>mt_version</c> column, so a mapping that ends up with
+/// both enabled used to surface as a raw duplicate-key <c>ArgumentException</c> (or invalid
+/// DDL with two <c>mt_version</c> columns) instead of an actionable configuration error.
+/// </summary>
+public class optimistic_concurrency_on_projection_target_fails_fast: BugIntegrationContext
+{
+    [Fact]
+    public void use_optimistic_concurrency_on_projected_document_throws_invalid_document_exception()
+    {
+        // ProjectionDocumentPolicy forces MyAggregate onto numeric revisions, then the fluent
+        // override runs after the policies and re-enables the Guid version — leaving both
+        // flavors on. That combination can never work and must fail fast — here already at
+        // store bootstrap, because the projection's ValidateConfiguration materializes the
+        // aggregate mapping.
+        var ex = Should.Throw<InvalidDocumentException>(() => StoreOptions(opts =>
+        {
+            opts.Projections.Add<AllGood>(ProjectionLifecycle.Inline);
+            opts.Schema.For<MyAggregate>().UseOptimisticConcurrency(true);
+        }));
+
+        ex.Message.ShouldContain("UseOptimisticConcurrency");
+        ex.Message.ShouldContain("UseNumericRevisions");
+        ex.Message.ShouldContain(nameof(MyAggregate));
+    }
+
+    [Fact]
+    public void switching_a_plain_document_from_numeric_revisions_to_optimistic_concurrency_also_fails_fast()
+    {
+        // Without any projection involved, stacking the two fluent calls leaves the revision
+        // metadata enabled from the first call while the second re-enables the Guid version.
+        StoreOptions(opts =>
+        {
+            opts.Schema.For<Target>().UseNumericRevisions(true);
+            opts.Schema.For<Target>().UseOptimisticConcurrency(true);
+        });
+
+        Should.Throw<InvalidDocumentException>(
+            () => theStore.Options.Storage.MappingFor(typeof(Target)));
+    }
+}

src/IssueService/StreamingMinimalEndpoints.cs

@@ -56,6 +56,18 @@ public static IEndpointRouteBuilder MapStreamingMinimalEndpoints(this IEndpointR
             (Guid id, IQuerySession session)
                 => new StreamOne<VersionlessDoc>(session.Query<VersionlessDoc>().Where(x => x.Id == id)));
 
+        // Projection-target document (numeric revisions forced by ProjectionDocumentPolicy) —
+        // served through StreamOne instead of StreamAggregate, the ETag is the numeric revision,
+        // which for a single-stream projection equals the source stream's version.
+        app.MapGet("/minimal/order-doc/{id:guid}",
+            (Guid id, IQuerySession session)
+                => new StreamOne<Order>(session.Query<Order>().Where(x => x.Id == id)));
+
+        // Plain document using numeric revisions via IRevisioned — no projection involved.
+        app.MapGet("/minimal/revisioned/{id:guid}",
+            (Guid id, IQuerySession session)
+                => new StreamOne<RevisionedIssueNote>(session.Query<RevisionedIssueNote>().Where(x => x.Id == id)));
+
         // --- StreamMany<T> ---
 
         app.MapGet("/minimal/issues/open",
@@ -173,11 +185,24 @@ public static IEndpointRouteBuilder MapStreamingMinimalEndpoints(this IEndpointR
 }
 
 /// <summary>
-/// A document type registered with version metadata disabled (no <c>mt_version</c> column),
-/// used to prove <see cref="StreamOne{T}"/> emits no ETag for versionless documents.
+/// A document type registered with version metadata disabled (no <c>mt_version</c> column of
+/// either flavor — Guid version or numeric revision), used to prove <see cref="StreamOne{T}"/>
+/// emits no ETag for versionless documents.
 /// </summary>
 public class VersionlessDoc
 {
     public Guid Id { get; set; }
     public string Name { get; set; }
 }
+
+/// <summary>
+/// A plain (non-projection) document using numeric revisions via
+/// <see cref="IRevisioned"/>, used to prove <see cref="StreamOne{T}"/>
+/// derives its ETag from the numeric revision.
+/// </summary>
+public class RevisionedIssueNote: IRevisioned
+{
+    public Guid Id { get; set; }
+    public string Name { get; set; }
+    public int Version { get; set; }
+}

src/Marten.AspNetCore.Testing/streaming_result_types_tests.cs

@@ -216,9 +216,11 @@ public async Task stream_one_suppresses_etag_when_emit_etag_is_false()
     [Fact]
     public async Task stream_one_emits_no_etag_when_version_metadata_disabled()
     {
-        // VersionlessDoc is registered with Metadata.Version.Enabled = false, so there is no
-        // mt_version column to derive an ETag from. EmitETag defaults to true, but the inline
-        // version read comes back null and no ETag (and no false 304) is produced.
+        // VersionlessDoc is registered with Metadata.Version.Enabled = false and carries no
+        // numeric revision metadata either (not a projection target, not IRevisioned), so there
+        // is no mt_version column of either flavor to derive an ETag from. EmitETag defaults to
+        // true, but the inline version read comes back null and no ETag (and no false 304) is
+        // produced.
         var doc = new VersionlessDoc { Id = Guid.NewGuid(), Name = "no version column" };
         await using (var session = theHost.Services.GetRequiredService<IDocumentStore>().LightweightSession())
         {
@@ -301,6 +303,124 @@ public void RecordSavedChanges(Marten.IDocumentSession session, Marten.Services.
         public void OnBeforeExecute(Npgsql.NpgsqlBatch batch) => Count++;
     }
 
+    // ──────────────── StreamOne<T> ETag — numeric-revision documents ────────────────
+
+    [Fact]
+    public async Task stream_one_emits_stream_version_etag_for_projection_target_document()
+    {
+        // Order is the target of an inline single-stream projection (Projections.Snapshot<Order>),
+        // so ProjectionDocumentPolicy forces numeric revisions and the projection writes the source
+        // stream's version into mt_version. Serving the projected document through StreamOne emits
+        // that revision as the ETag — the same value StreamAggregate derives for the same stream.
+        var orderId = Guid.NewGuid();
+        await using (var session = theHost.Services.GetRequiredService<IDocumentStore>().LightweightSession())
+        {
+            session.Events.StartStream<Order>(orderId, new OrderPlaced("Projected Book", 12.50m));
+            await session.SaveChangesAsync();
+        }
+
+        var result = await theHost.Scenario(s =>
+        {
+            s.Get.Url($"/minimal/order-doc/{orderId}");
+            s.StatusCodeShouldBe(200);
+            s.ContentTypeShouldBe("application/json");
+        });
+
+        result.Context.Response.Headers["ETag"].ToString().ShouldBe("\"1\"");
+
+        var order = result.ReadAsJson<Order>();
+        order.Id.ShouldBe(orderId);
+        order.Description.ShouldBe("Projected Book");
+    }
+
+    [Fact]
+    public async Task stream_one_returns_304_when_if_none_match_matches_projection_revision()
+    {
+        var orderId = Guid.NewGuid();
+        await using (var session = theHost.Services.GetRequiredService<IDocumentStore>().LightweightSession())
+        {
+            session.Events.StartStream<Order>(orderId, new OrderPlaced("Cached Projection", 3.00m));
+            await session.SaveChangesAsync();
+        }
+
+        var second = await theHost.Scenario(s =>
+        {
+            s.Get.Url($"/minimal/order-doc/{orderId}");
+            s.WithRequestHeader("If-None-Match", "\"1\"");
+            s.StatusCodeShouldBe(304);
+        });
+
+        second.Context.Response.Headers["ETag"].ToString().ShouldBe("\"1\"");
+        second.ReadAsText().ShouldBeNullOrEmpty();
+    }
+
+    [Fact]
+    public async Task stream_one_projection_etag_changes_when_the_stream_advances()
+    {
+        var orderId = Guid.NewGuid();
+        await using (var session = theHost.Services.GetRequiredService<IDocumentStore>().LightweightSession())
+        {
+            session.Events.StartStream<Order>(orderId, new OrderPlaced("Evolving Book", 8.00m));
+            await session.SaveChangesAsync();
+        }
+
+        var first = await theHost.Scenario(s =>
+        {
+            s.Get.Url($"/minimal/order-doc/{orderId}");
+            s.StatusCodeShouldBe(200);
+        });
+        first.Context.Response.Headers["ETag"].ToString().ShouldBe("\"1\"");
+
+        await using (var session = theHost.Services.GetRequiredService<IDocumentStore>().LightweightSession())
+        {
+            session.Events.Append(orderId, new OrderShipped());
+            await session.SaveChangesAsync();
+        }
+
+        // The previously-cached ETag is now stale: full body again, with the new revision.
+        var second = await theHost.Scenario(s =>
+        {
+            s.Get.Url($"/minimal/order-doc/{orderId}");
+            s.WithRequestHeader("If-None-Match", "\"1\"");
+            s.StatusCodeShouldBe(200);
+        });
+
+        second.Context.Response.Headers["ETag"].ToString().ShouldBe("\"2\"");
+
+        var order = second.ReadAsJson<Order>();
+        order.Shipped.ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task stream_one_emits_revision_etag_for_plain_revisioned_document()
+    {
+        // Not a projection target: RevisionedIssueNote opts into numeric revisions by
+        // implementing IRevisioned, which also gives its mt_version column the narrower
+        // integer width (#4614) — proving the revision read handles both column widths.
+        var note = new RevisionedIssueNote { Id = Guid.NewGuid(), Name = "rev one" };
+        await using (var session = theHost.Services.GetRequiredService<IDocumentStore>().LightweightSession())
+        {
+            session.Store(note);
+            await session.SaveChangesAsync();
+        }
+
+        var result = await theHost.Scenario(s =>
+        {
+            s.Get.Url($"/minimal/revisioned/{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/revisioned/{note.Id}");
+            s.WithRequestHeader("If-None-Match", "\"1\"");
+            s.StatusCodeShouldBe(304);
+        });
+    }
+
     // ───────────────────────── StreamMany<T> ─────────────────────────
 
     [Fact]

src/Marten.AspNetCore/QueryableExtensions.cs

@@ -19,9 +19,12 @@ public static class QueryableExtensions
     /// When <paramref name="emitETag"/> is true (the default), the document's <c>mt_version</c>
     /// is read <b>inline with the document in the same single round trip</b> (piggy-backed onto the
     /// streaming query, analogous to the <c>count(*) OVER()</c> stats column) and written as a quoted
-    /// <c>ETag</c> response header. If the incoming request's <c>If-None-Match</c> header matches that
-    /// version, a <c>304 Not Modified</c> is written instead, with an empty body. Document types with
-    /// version metadata disabled (no <c>mt_version</c> column) emit no ETag.
+    /// <c>ETag</c> response header — a quoted GUID under Guid optimistic concurrency, or a quoted
+    /// integer for numeric-revision documents (projection targets, <c>IRevisioned</c> types), where
+    /// a <c>SingleStreamProjection</c> target's revision is the source stream's version. If the
+    /// incoming request's <c>If-None-Match</c> header matches that value, a <c>304 Not Modified</c>
+    /// is written instead, with an empty body. Document types with neither version nor revision
+    /// metadata enabled (no <c>mt_version</c> column) emit no ETag.
     /// </para>
     /// </summary>
     /// <param name="queryable"></param>
@@ -68,10 +71,14 @@ public static async Task WriteSingle<T>(
             return;
         }
 
-        if (result.Version.HasValue)
-        {
-            var etag = ETagHelpers.Format(result.Version.Value);
+        var etag = result.Version.HasValue
+            ? ETagHelpers.Format(result.Version.Value)
+            : result.Revision.HasValue
+                ? ETagHelpers.Format(result.Revision.Value)
+                : null;
 
+        if (etag != null)
+        {
             if (ETagHelpers.IfNoneMatchMatches(context, etag))
             {
                 context.Response.StatusCode = StatusCodes.Status304NotModified;

src/Marten.AspNetCore/StreamOne.cs

@@ -53,10 +53,11 @@ public StreamOne(IQueryable<T> queryable)
 
     /// <summary>
     /// Whether to emit an <c>ETag</c> response header derived from the document's
-    /// <c>mt_version</c>, and honor an incoming <c>If-None-Match</c> request header by
-    /// responding <c>304 Not Modified</c> with an empty body when it matches. Defaults to
-    /// <c>true</c>. Set to <c>false</c> to opt out if a consumer's contract cannot tolerate
-    /// the extra header.
+    /// <c>mt_version</c> (a quoted GUID under Guid optimistic concurrency, a quoted integer
+    /// for numeric-revision documents such as projection targets), and honor an incoming
+    /// <c>If-None-Match</c> request header by responding <c>304 Not Modified</c> with an
+    /// empty body when it matches. Defaults to <c>true</c>. Set to <c>false</c> to opt out
+    /// if a consumer's contract cannot tolerate the extra header.
     /// </summary>
     public bool EmitETag { get; init; } = true;
 

src/Marten/Internal/Sessions/QuerySession.Execution.cs

@@ -102,6 +102,21 @@ internal async Task<bool> StreamOne(DbCommand command, Stream stream, Cancellati
         }
     }
 
+    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);
+        }
+        finally
+        {
+            await reader.CloseAsync().ConfigureAwait(false);
+        }
+    }
+
     internal async Task<int> StreamMany(DbCommand command, Stream stream, CancellationToken token)
     {
         await using var reader = await ExecuteReaderAsync(command, token).ConfigureAwait(false);