diff --git a/src/Marten/Events/Daemon/Internals/ProjectionUpdateBatch.cs b/src/Marten/Events/Daemon/Internals/ProjectionUpdateBatch.cs index 7b5914df60..2d764cc56c 100644 --- a/src/Marten/Events/Daemon/Internals/ProjectionUpdateBatch.cs +++ b/src/Marten/Events/Daemon/Internals/ProjectionUpdateBatch.cs @@ -346,6 +346,15 @@ public async Task WaitForCompletion() { stream.PrepareEvents(0, ((IMartenSession)_session).Options.EventGraph, new Queue(), _session); + // #5062: an Append side effect that ended up with no events has nothing to + // write, and calling mt_quick_append_events with empty arrays is a wasted + // round trip at best. A StartStream still needs its mt_streams row, so only + // the append shape is skipped. + if (stream.ActionType != StreamActionType.Start && !stream.Events.Any()) + { + continue; + } + var op = stream.ActionType == StreamActionType.Start ? eventStorage.InsertStream(stream) : eventStorage.QuickAppendEvents(stream); applyOperation(op); } diff --git a/src/Marten/Events/Operations/QuickAppendEventsOperationBase.cs b/src/Marten/Events/Operations/QuickAppendEventsOperationBase.cs index a4115bcd4c..723bfa526e 100644 --- a/src/Marten/Events/Operations/QuickAppendEventsOperationBase.cs +++ b/src/Marten/Events/Operations/QuickAppendEventsOperationBase.cs @@ -322,6 +322,23 @@ public async Task PostprocessAsync(DbDataReader reader, IList excepti { if (await reader.ReadAsync(token).ConfigureAwait(false)) { + // #5062: nothing to post-process for an append with no events -- but the row + // still has to be consumed above so the reader stays aligned for the rest of + // the page. Bail before touching field 0: on a database whose + // mt_quick_append_events predates the COALESCE fix, the empty case comes back + // as ARRAY[NULL] and GetFieldValueAsync throws InvalidCastException, + // which would then replace whatever exception actually made the caller fail. + // #5062: nothing to post-process for an append with no events -- but the row + // still has to be consumed above so the reader stays aligned for the rest of + // the page. Bail before touching field 0: on a database whose + // mt_quick_append_events predates the COALESCE fix, the empty case comes back + // as ARRAY[NULL] and GetFieldValueAsync throws InvalidCastException, + // which would then replace whatever exception actually made the caller fail. + if (Stream.Events.Count == 0) + { + return; + } + var values = await reader.GetFieldValueAsync(0, token).ConfigureAwait(false); var finalVersion = values[0]; diff --git a/src/Marten/Events/Schema/QuickAppendEventFunction.cs b/src/Marten/Events/Schema/QuickAppendEventFunction.cs index 9d1d42c829..b5be4b7f1f 100644 --- a/src/Marten/Events/Schema/QuickAppendEventFunction.cs +++ b/src/Marten/Events/Schema/QuickAppendEventFunction.cs @@ -229,7 +229,13 @@ public override void WriteCreateStatement(Migrator migrator, TextWriter writer) end if; index := 1; - return_value := ARRAY[event_version + array_length(event_ids, 1)]; + -- #5062: array_length('{{}}', 1) is NULL in PostgreSQL, not 0, so a call with an + -- empty event array used to return ARRAY[NULL] -- a bigint[] whose only element + -- is NULL, which Npgsql cannot read into long[] ('Cannot read a non-nullable + -- collection of elements because the returned array contains nulls'). COALESCE + -- makes the empty case mean what it says: zero events appended, so the final + -- version is the stream's current version. + return_value := ARRAY[event_version + COALESCE(array_length(event_ids, 1), 0)]; foreach event_id in ARRAY event_ids loop diff --git a/src/TenantPartitionedEventsTests/Regressions/Bug_5062_empty_quick_append.cs b/src/TenantPartitionedEventsTests/Regressions/Bug_5062_empty_quick_append.cs new file mode 100644 index 0000000000..ff8fedb414 --- /dev/null +++ b/src/TenantPartitionedEventsTests/Regressions/Bug_5062_empty_quick_append.cs @@ -0,0 +1,165 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Core; +using JasperFx.Events; +using Marten.Events.Operations; +using Marten.Internal.Sessions; +using Marten.Testing.Harness; +using Npgsql; +using NpgsqlTypes; +using Shouldly; +using TenantPartitionedEventsTests.Fixtures; +using Weasel.Postgresql; +using Xunit; + +namespace TenantPartitionedEventsTests.Regressions; + +/// +/// #5062 — mt_quick_append_events called with an EMPTY event array used to return +/// {NULL}, because array_length('{}', 1) is NULL in PostgreSQL (not 0), so +/// event_version + array_length(event_ids, 1) is NULL. Npgsql then failed to read that +/// bigint[] into long[] with +/// InvalidCastException: Cannot read a non-nullable collection of elements because the +/// returned array contains nulls. +/// +/// +/// The cast failure surfaced from , +/// i.e. from inside the batch's callback loop — so it propagated out of the loop and discarded +/// whatever exception had already been collected for the batch. Callers saw an unrelated, +/// non-retryable InvalidCastException instead of the real error. +/// +/// +[Collection("guid-partitioned")] +public class Bug_5062_empty_quick_append +{ + private readonly GuidPartitionedFixture _fixture; + + public Bug_5062_empty_quick_append(GuidPartitionedFixture fixture) + { + _fixture = fixture; + } + + /// + /// The issue's SQL repro. The parameter list of mt_quick_append_events varies with + /// configuration (metadata columns, server timestamps, tag tables, bigint events), so read + /// the deployed signature back out of the catalog and feed every array parameter an empty + /// array — that way this pins the function rather than one config's hand-written call. + /// + [Fact] + public async Task function_returns_a_non_null_version_for_an_empty_event_array() + { + var tenant = PartitionedFixtureBase.NewTenant(); + await _fixture.Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); + + var streamId = await _fixture.AppendNEventsAsync(tenant, 3); + + await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync(); + + var arguments = await readFunctionArgumentsAsync(conn); + var sql = $"select {_fixture.SchemaName}.mt_quick_append_events({buildEmptyArgumentList(arguments)})"; + + await using var cmd = conn.CreateCommand(sql) + .With("stream", streamId, NpgsqlDbType.Uuid) + .With("tenant", tenant, NpgsqlDbType.Varchar); + + var raw = await cmd.ExecuteScalarAsync(); + + // int[] when EnableBigIntEvents is off, bigint[] when it is on. Before the fix this came + // back as a single-element array holding NULL. + var values = raw switch + { + int[] ints => Array.ConvertAll(ints, x => (long)x), + long[] longs => longs, + _ => throw new InvalidOperationException($"Unexpected return type {raw?.GetType()?.FullName}") + }; + + values.Length.ShouldBe(1); + + // Zero events appended -> the stream's version is unchanged. + values[0].ShouldBe(3); + } + + /// + /// Same empty append, but driven through Marten's own generated call site + result read, which + /// is where the InvalidCastException actually landed on the reporter's system. + /// + [Fact] + public async Task empty_quick_append_operation_is_a_clean_no_op() + { + var tenant = PartitionedFixtureBase.NewTenant(); + await _fixture.Store.Advanced.AddMartenManagedTenantsAsync(CancellationToken.None, tenant); + + var streamId = await _fixture.AppendNEventsAsync(tenant, 2); + var eventCountBefore = await _fixture.CountEventsForTenantAsync(tenant, _fixture.SchemaName); + + await using var session = (DocumentSessionBase)_fixture.Store.LightweightSession(tenant); + + var stream = StreamAction.Append(streamId, Array.Empty()); + stream.TenantId = tenant; + + var op = (QuickAppendEventsOperationBase)session.EventStorage().QuickAppendEvents(stream); + op.Events = _fixture.Store.Options.EventGraph; + session.QueueOperation(op); + + await Should.NotThrowAsync(() => session.SaveChangesAsync()); + + (await _fixture.CountEventsForTenantAsync(tenant, _fixture.SchemaName)).ShouldBe(eventCountBefore); + + await using var query = _fixture.Store.QuerySession(tenant); + var state = await query.Events.FetchStreamStateAsync(streamId); + state.ShouldNotBeNull(); + state.Version.ShouldBe(2); + } + + private async Task> readFunctionArgumentsAsync(NpgsqlConnection conn) + { + var raw = (string?)await conn.CreateCommand(@" +select pg_get_function_arguments(p.oid) +from pg_proc p +join pg_namespace n on n.oid = p.pronamespace +where n.nspname = :schema and p.proname = 'mt_quick_append_events'") + .With("schema", _fixture.SchemaName, NpgsqlDbType.Varchar) + .ExecuteScalarAsync(); + + raw.ShouldNotBeNull(); + + // No PostgreSQL type rendered in this signature contains a comma ("character varying[]", + // "timestamp with time zone[]", "integer DEFAULT NULL::integer"), so a flat split is safe. + return raw!.Split(", ", StringSplitOptions.RemoveEmptyEntries).ToArray(); + } + + private static string buildEmptyArgumentList(IReadOnlyList arguments) + { + var rendered = new List(arguments.Count); + + for (var i = 0; i < arguments.Count; i++) + { + // " " or " DEFAULT " + var declaration = arguments[i]; + var defaultAt = declaration.IndexOf(" DEFAULT ", StringComparison.Ordinal); + if (defaultAt > -1) + { + declaration = declaration.Substring(0, defaultAt); + } + + var type = declaration.Substring(declaration.IndexOf(' ') + 1); + + rendered.Add(i switch + { + 0 => ":stream", + 1 => "'Trip'", + 2 => ":tenant", + // Every remaining event column is an array; the lone scalar is the trailing + // expected_version, which NULL turns into "no optimistic concurrency check". + _ => type.EndsWith("[]", StringComparison.Ordinal) ? $"'{{}}'::{type}" : $"NULL::{type}" + }); + } + + return rendered.Join(", "); + } +}