From 1d5ded4884231454092b1bfe7bdbd3cdb9b08dc0 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 2 Aug 2026 13:03:42 -0500 Subject: [PATCH] feat: pluggable binary event serialization (IEventBinarySerializer) (#388) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parity with Marten's IEventBinarySerializer (marten#4515, shipped 9.20.2). CritterWatch ships Marten/PostgreSQL and Polecat/SQL Server flavors from one source tree, and CritterWatch#896 has already opted its highest-volume internal event into binary on the Marten side — measured at field scale as -97.9% on the wire, -60.4% on disk, -73.4% append p50. Until now the SQL Server flavor kept paying the JSON cost the Marten one shed. The API mirrors Marten's exactly so a store-agnostic consumer can wire either: opts.Events.UseBinarySerializer(serializer); // explicit, per event type [BinaryEvent] + opts.Events.DefaultBinarySerializer; // attribute-driven The part worth copying is the coexistence design, and that is what this follows: an additive nullable `bdata varbinary(max)` column beside `data`, with `bdata IS NULL` as the per-ROW discriminator. JSON and binary rows live in the same pc_events table, so the feature switches on for an existing store with no migration of existing event data and switches back off just as safely. The column is added unconditionally (not only when a serializer is configured) so the read projection has one fixed shape; Weasel's additive delta adds it on the next schema apply, and pre-existing rows read through the JSON path untouched. Quick mode was in scope from the start rather than as a phase 2 — Polecat is QuickAppend-only, so there is no Rich-mode split to sequence and the CritterWatch store's configuration is covered by the first pass. Notable implementation points: - Every read path dispatches, not just the main one. Polecat has four independent readers over pc_events (PcEventsRowReader, the two DCB tag-query readers in EventOperations, the LINQ EventListHandler, and the daemon's PolecatEventLoader), each with its own SELECT and its own deserialization. bdata is pinned at ordinal 10 — after the previously-locked 0-9 block, before the optional metadata columns, which shift to 11 — so the dispatch reads a stable ordinal everywhere. The two EventOperations projections that duplicated the canonical column list by hand now compose it from PcEventsRowReader instead; three hand-written copies of one list is how they stop agreeing. - Masking rewrites bdata, not just data. OverwriteEventOperation (ApplyEventDataMasking) previously wrote only the JSON column. For a binary event that would have left the original, unmasked payload readable in bdata — for a GDPR masking operation, the entire point missed. Same for CompletelyReplaceEvent and stream compaction, where the REPLACEMENT body's type decides the row's format and a stale bdata would keep being read. - A [BinaryEvent] type with no serializer configured throws instead of silently writing JSON, and the append path deliberately does not short-circuit the resolve to make that cheaper: a store that quietly ignored the attribute would have write-amplification characteristics that do not match its configuration, which is the problem the feature exists to solve. The resolve is a per-event-type cached dictionary hit. 12 tests covering the wire format (payload in bdata, '{}' placeholder in data), JSON/binary coexistence within a single stream, reading rows written before the serializer was configured, attribute vs explicit-registration precedence, the misconfiguration throw, and the dispatch through all of inline projections, the async daemon, the event LINQ provider, and masking. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01G8tN8ApXiKhyVzia4iwmof --- .../binary_event_serialization_tests.cs | 421 ++++++++++++++++++ src/Polecat/Events/BinaryEventAttribute.cs | 28 ++ .../Events/Daemon/PolecatEventLoader.cs | 6 +- src/Polecat/Events/EventGraph.cs | 112 +++++ src/Polecat/Events/EventOperations.cs | 45 +- src/Polecat/Events/IEventBinarySerializer.cs | 42 ++ .../Events/Internal/PcEventsRowReader.cs | 25 +- .../Events/Linq/EventLinqQueryProvider.cs | 2 +- src/Polecat/Events/Linq/EventListHandler.cs | 9 +- .../Protected/OverwriteEventOperation.cs | 39 +- .../Events/Protected/ReplaceEventOperation.cs | 18 +- .../Events/Protected/StreamCompacting.cs | 16 +- src/Polecat/Events/Schema/EventsTable.cs | 11 + .../PolecatQuickAppendEventsOperation.cs | 14 +- .../Storage/SqlServerEventStoreDialect.cs | 13 +- .../Storage/SqlServerStorageDialect.cs | 2 + src/Polecat/StoreOptions.cs | 26 ++ 17 files changed, 777 insertions(+), 52 deletions(-) create mode 100644 src/Polecat.Tests/Events/binary_event_serialization_tests.cs create mode 100644 src/Polecat/Events/BinaryEventAttribute.cs create mode 100644 src/Polecat/Events/IEventBinarySerializer.cs diff --git a/src/Polecat.Tests/Events/binary_event_serialization_tests.cs b/src/Polecat.Tests/Events/binary_event_serialization_tests.cs new file mode 100644 index 00000000..65976c6f --- /dev/null +++ b/src/Polecat.Tests/Events/binary_event_serialization_tests.cs @@ -0,0 +1,421 @@ +using System.Text; +using System.Text.Json; +using JasperFx.Events; +using JasperFx.Events.Projections; +using Polecat.Events; +using Polecat.Linq; +using Polecat.Projections; +using Polecat.Tests.Harness; +using Shouldly; + +namespace Polecat.Tests.Events; + +#region sample_polecat_binary_event_serializer + +/// +/// A deliberately non-JSON binary format, so a test can prove a row really did travel through the +/// binary path: the bytes are a length-prefixed UTF-8 blob that no JSON parser would accept, and a +/// four-byte magic header lets the assertions recognize it on the wire. +/// +public sealed class TestBinaryEventSerializer : IEventBinarySerializer +{ + public static readonly byte[] Magic = "PCB1"u8.ToArray(); + + /// How many times Serialize/Deserialize were called — proves the path was taken. + public int SerializeCount; + + public int DeserializeCount; + + public byte[] Serialize(Type type, object data) + { + Interlocked.Increment(ref SerializeCount); + var payload = JsonSerializer.SerializeToUtf8Bytes(data, type); + var buffer = new byte[Magic.Length + payload.Length]; + Magic.CopyTo(buffer, 0); + payload.CopyTo(buffer, Magic.Length); + return buffer; + } + + public object Deserialize(Type type, byte[] data) + { + Interlocked.Increment(ref DeserializeCount); + if (data.Length < Magic.Length || !data.AsSpan(0, Magic.Length).SequenceEqual(Magic)) + { + throw new InvalidOperationException("Not a payload written by this serializer."); + } + + return JsonSerializer.Deserialize(data.AsSpan(Magic.Length), type)!; + } +} + +#endregion + +public record BinaryPayloadRecorded(string Name, int Amount); + +public record JsonPayloadRecorded(string Name, int Amount); + +[BinaryEvent] +public record AttributeMarkedRecorded(string Name); + +public class BinaryLedger +{ + public Guid Id { get; set; } + public int Total { get; set; } + public List Names { get; set; } = new(); +} + +public partial class BinaryLedgerProjection : SingleStreamProjection +{ + public void Apply(BinaryLedger ledger, BinaryPayloadRecorded e) + { + ledger.Total += e.Amount; + ledger.Names.Add(e.Name); + } + + public void Apply(BinaryLedger ledger, JsonPayloadRecorded e) + { + ledger.Total += e.Amount; + ledger.Names.Add(e.Name); + } +} + +/// +/// polecat#388: pluggable binary event serialization at parity with Marten's +/// IEventBinarySerializer (marten#4515). The design point under test throughout is the +/// coexistence one: an additive nullable bdata column with bdata IS NULL as +/// the per-row discriminator, so JSON and binary events live in the same pc_events table +/// and the feature is switchable on an existing store with no data migration. +/// +public class binary_event_serialization_tests : OneOffConfigurationsContext +{ + private readonly TestBinaryEventSerializer _serializer = new(); + + private async Task ConfigureAndApply(Action configure) + { + ConfigureStore(configure); + await theDatabase.ApplyAllConfiguredChangesToDatabaseAsync(); + } + + private Task ConfigureBinaryStore(Action? extra = null) => ConfigureAndApply(opts => + { + opts.Events.UseBinarySerializer(_serializer); + extra?.Invoke(opts); + }); + + private async Task<(string data, byte[]? bdata)> ReadRawRowAsync(long sequence) + { + await using var conn = await OpenConnectionAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"SELECT CONVERT(nvarchar(max), data), bdata FROM {theStore.Options.EventGraph.EventsTableName} WHERE seq_id = @seq"; + var p = cmd.CreateParameter(); + p.ParameterName = "@seq"; + p.Value = sequence; + cmd.Parameters.Add(p); + + await using var reader = await cmd.ExecuteReaderAsync(TestContext.Current.CancellationToken); + (await reader.ReadAsync(TestContext.Current.CancellationToken)).ShouldBeTrue(); + var data = reader.GetString(0); + var bdata = reader.IsDBNull(1) ? null : (byte[])reader.GetValue(1); + return (data, bdata); + } + + [Fact] + public async Task pc_events_carries_a_nullable_bdata_column() + { + await ConfigureAndApply(_ => { }); + + await using var conn = await OpenConnectionAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = """ + SELECT c.is_nullable, TYPE_NAME(c.system_type_id), c.max_length + FROM sys.columns c + WHERE c.object_id = OBJECT_ID(@table) AND c.name = 'bdata' + """; + var p = cmd.CreateParameter(); + p.ParameterName = "@table"; + p.Value = theStore.Options.EventGraph.EventsTableName; + cmd.Parameters.Add(p); + + await using var reader = await cmd.ExecuteReaderAsync(TestContext.Current.CancellationToken); + (await reader.ReadAsync(TestContext.Current.CancellationToken)) + .ShouldBeTrue("pc_events should carry the bdata column even with no serializer configured"); + reader.GetBoolean(0).ShouldBeTrue("bdata must be nullable — NULL is the JSON-row discriminator"); + reader.GetString(1).ShouldBe("varbinary"); + reader.GetInt16(2).ShouldBe((short)-1); // varbinary(max) + } + + [Fact] + public async Task an_unconfigured_store_writes_json_and_leaves_bdata_null() + { + await ConfigureAndApply(_ => { }); + + await using var session = theStore.LightweightSession(); + var streamId = session.Events.StartStream(new JsonPayloadRecorded("plain", 3)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + + var events = await session.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + var (data, bdata) = await ReadRawRowAsync(events[0].Sequence); + + bdata.ShouldBeNull(); + data.ShouldContain("plain"); + } + + [Fact] + public async Task a_binary_event_round_trips_through_bdata_and_leaves_data_as_a_placeholder() + { + await ConfigureBinaryStore(); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream(new BinaryPayloadRecorded("binary", 7)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + _serializer.SerializeCount.ShouldBe(1); + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events.Count.ShouldBe(1); + events[0].Data.ShouldBeOfType().ShouldBe(new BinaryPayloadRecorded("binary", 7)); + _serializer.DeserializeCount.ShouldBeGreaterThan(0); + + // On the wire: the payload is in bdata, and `data` holds only the placeholder — so the + // saving the feature exists for is real, not just a round trip that happens to work. + var (data, bdata) = await ReadRawRowAsync(events[0].Sequence); + bdata.ShouldNotBeNull(); + bdata!.AsSpan(0, TestBinaryEventSerializer.Magic.Length).SequenceEqual(TestBinaryEventSerializer.Magic) + .ShouldBeTrue(); + Encoding.UTF8.GetString(bdata).ShouldContain("binary"); + data.ShouldBe("{}"); + data.ShouldNotContain("binary"); + } + + [Fact] + public async Task json_and_binary_events_coexist_in_one_stream() + { + // The coexistence property is the whole reason this design is low-risk to adopt: only the + // opted-in type changes format, and per-ROW dispatch means one stream can hold both. + await ConfigureBinaryStore(); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new JsonPayloadRecorded("first", 1), + new BinaryPayloadRecorded("second", 2), + new JsonPayloadRecorded("third", 3)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events.Count.ShouldBe(3); + events.Select(x => x.Data).ShouldBe([ + new JsonPayloadRecorded("first", 1), + new BinaryPayloadRecorded("second", 2), + new JsonPayloadRecorded("third", 3) + ]); + + (await ReadRawRowAsync(events[0].Sequence)).bdata.ShouldBeNull(); + (await ReadRawRowAsync(events[1].Sequence)).bdata.ShouldNotBeNull(); + (await ReadRawRowAsync(events[2].Sequence)).bdata.ShouldBeNull(); + } + + [Fact] + public async Task rows_written_before_the_serializer_was_configured_still_read_as_json() + { + // The "switch it on for an existing store with no migration" claim, exercised directly: + // append as JSON, then reconfigure the same schema with the serializer registered and read + // the pre-existing row back. + await ConfigureAndApply(_ => { }); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream(new BinaryPayloadRecorded("legacy-json", 5)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + // Reconfigure WITHOUT dropping the schema — same tables, now with binary registered. + ConfigureStore(opts => opts.Events.UseBinarySerializer(_serializer)); + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events[0].Data.ShouldBe(new BinaryPayloadRecorded("legacy-json", 5)); + _serializer.DeserializeCount.ShouldBe(0, "a bdata IS NULL row must not go through the binary path"); + + // And the next append for that type does use binary — the two formats sit side by side. + await using (var session = theStore.LightweightSession()) + { + session.Events.Append(streamId, new BinaryPayloadRecorded("now-binary", 6)); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query2 = theStore.QuerySession(); + var after = await query2.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + after.Count.ShouldBe(2); + after.Select(x => ((BinaryPayloadRecorded)x.Data).Name).ShouldBe(["legacy-json", "now-binary"]); + (await ReadRawRowAsync(after[0].Sequence)).bdata.ShouldBeNull(); + (await ReadRawRowAsync(after[1].Sequence)).bdata.ShouldNotBeNull(); + } + + [Fact] + public async Task the_binary_event_attribute_resolves_against_the_default_serializer() + { + await ConfigureAndApply(opts => opts.Events.DefaultBinarySerializer = _serializer); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new AttributeMarkedRecorded("attributed"), + new JsonPayloadRecorded("not-attributed", 1)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events[0].Data.ShouldBe(new AttributeMarkedRecorded("attributed")); + events[1].Data.ShouldBe(new JsonPayloadRecorded("not-attributed", 1)); + + (await ReadRawRowAsync(events[0].Sequence)).bdata.ShouldNotBeNull(); + (await ReadRawRowAsync(events[1].Sequence)).bdata + .ShouldBeNull("[BinaryEvent] is per type — an unmarked type stays on the JSON path"); + } + + [Fact] + public async Task an_explicit_registration_wins_over_the_attribute_and_the_default() + { + var explicitSerializer = new TestBinaryEventSerializer(); + await ConfigureAndApply(opts => + { + opts.Events.DefaultBinarySerializer = _serializer; + opts.Events.UseBinarySerializer(explicitSerializer); + }); + + await using var session = theStore.LightweightSession(); + session.Events.StartStream(new AttributeMarkedRecorded("explicit")); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + + explicitSerializer.SerializeCount.ShouldBe(1); + _serializer.SerializeCount.ShouldBe(0); + } + + [Fact] + public async Task a_marked_type_with_no_serializer_configured_throws_rather_than_silently_writing_json() + { + // A silent fallback would leave a store whose write amplification does not match its + // configuration — which is exactly the problem the feature exists to fix. + await ConfigureAndApply(_ => { }); + + await using var session = theStore.LightweightSession(); + session.Events.StartStream(new AttributeMarkedRecorded("unconfigured")); + + var ex = await Should.ThrowAsync( + async () => await session.SaveChangesAsync(TestContext.Current.CancellationToken)); + ex.Message.ShouldContain("[BinaryEvent]"); + ex.Message.ShouldContain("DefaultBinarySerializer"); + } + + [Fact] + public async Task binary_events_flow_through_an_inline_projection() + { + await ConfigureBinaryStore(opts => + opts.Projections.Add(ProjectionLifecycle.Inline)); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new BinaryPayloadRecorded("b", 10), + new JsonPayloadRecorded("j", 5)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var ledger = await query.LoadAsync(streamId, TestContext.Current.CancellationToken); + ledger.ShouldNotBeNull(); + ledger!.Total.ShouldBe(15); + ledger.Names.ShouldBe(["b", "j"]); + } + + [Fact] + public async Task binary_events_flow_through_an_async_projection_via_the_daemon() + { + // The daemon's event loader is a separate reader from FetchStreamAsync, with its own SELECT + // and its own deserialization — so it needs its own coverage for the bdata dispatch. + await ConfigureBinaryStore(opts => + opts.Projections.Add(ProjectionLifecycle.Async)); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new BinaryPayloadRecorded("b1", 4), + new BinaryPayloadRecorded("b2", 6), + new JsonPayloadRecorded("j1", 1)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + using var daemon = await theStore.BuildProjectionDaemonAsync(); + await daemon.StartAllAsync(); + await daemon.WaitForNonStaleData(TimeSpan.FromSeconds(30)); + + await using var query = theStore.QuerySession(); + var ledger = await query.LoadAsync(streamId, TestContext.Current.CancellationToken); + ledger.ShouldNotBeNull(); + ledger!.Total.ShouldBe(11); + ledger.Names.ShouldBe(["b1", "b2", "j1"]); + } + + [Fact] + public async Task binary_events_hydrate_through_the_event_linq_provider() + { + // A third distinct reader (EventListHandler) — same dispatch, separately covered. + await ConfigureBinaryStore(); + + await using (var session = theStore.LightweightSession()) + { + session.Events.StartStream(new BinaryPayloadRecorded("via-linq", 42)); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var events = await query.Events.QueryAllRawEvents() + .Where(e => e.EventTypeName == "binary_payload_recorded") + .ToListAsync(TestContext.Current.CancellationToken); + + events.Count.ShouldBe(1); + events[0].Data.ShouldBe(new BinaryPayloadRecorded("via-linq", 42)); + } + + [Fact] + public async Task masking_a_binary_event_rewrites_bdata_not_just_the_json_column() + { + // If masking only rewrote `data`, the original payload would stay readable in bdata — for a + // GDPR masking operation that is the entire point missed. + await ConfigureBinaryStore(); + theStore.Events.AddMaskingRuleForProtectedInformation( + e => e with { Name = "****" }); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream(new BinaryPayloadRecorded("secret", 9)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await theStore.Advanced.ApplyEventDataMasking(x => x.IncludeStream(streamId), + TestContext.Current.CancellationToken); + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events[0].Data.ShouldBeOfType().Name.ShouldBe("****"); + + var (data, bdata) = await ReadRawRowAsync(events[0].Sequence); + bdata.ShouldNotBeNull("the masked row must stay binary"); + Encoding.UTF8.GetString(bdata!).ShouldNotContain("secret"); + data.ShouldBe("{}"); + } +} diff --git a/src/Polecat/Events/BinaryEventAttribute.cs b/src/Polecat/Events/BinaryEventAttribute.cs new file mode 100644 index 00000000..68a5cf74 --- /dev/null +++ b/src/Polecat/Events/BinaryEventAttribute.cs @@ -0,0 +1,28 @@ +namespace Polecat.Events; + +/// +/// Marks an event type as binary-serialized: its pc_events.data column holds the +/// '{}' placeholder and the real payload lives in bdata, written and read by an +/// . Mirrors Marten's [BinaryEvent] +/// (); tracked as +/// polecat#388. +/// +/// +/// +/// The serializer for an attribute-marked type is the store-wide +/// opts.Events.DefaultBinarySerializer. An explicit +/// opts.Events.UseBinarySerializer<TEvent>(serializer) registration takes +/// precedence. If a type is attribute-marked but neither is configured, the store throws when +/// it first resolves that event type rather than silently writing JSON — a silent fallback +/// would produce a store whose write amplification quietly does not match its configuration. +/// +/// +/// JSON and binary events coexist per event type in the same table, so applying this to a +/// single event type is a safe in-place change: existing JSON rows keep bdata = NULL +/// and keep reading through the JSON path. +/// +/// +[AttributeUsage(AttributeTargets.Class | AttributeTargets.Struct, Inherited = false)] +public sealed class BinaryEventAttribute : Attribute +{ +} diff --git a/src/Polecat/Events/Daemon/PolecatEventLoader.cs b/src/Polecat/Events/Daemon/PolecatEventLoader.cs index e7c35afe..1def9708 100644 --- a/src/Polecat/Events/Daemon/PolecatEventLoader.cs +++ b/src/Polecat/Events/Daemon/PolecatEventLoader.cs @@ -80,7 +80,7 @@ private async Task LoadInternalAsync(EventRequest request, Cancellati await using var cmd = conn.CreateCommand(); cmd.CommandText = $""" SELECT TOP(@batchSize) seq_id, id, stream_id, version, data, type, timestamp, - tenant_id, dotnet_type, is_archived + tenant_id, dotnet_type, is_archived, bdata FROM {_events.EventsTableName} WHERE seq_id > @floor AND seq_id <= @ceiling AND is_archived = 0{tenantPredicate} ORDER BY seq_id; @@ -106,6 +106,8 @@ SELECT TOP(@batchSize) seq_id, id, stream_id, version, data, type, timestamp, var tenantId = reader.GetString(7); var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8); var isArchived = reader.GetBoolean(9); + // #388: non-null bdata means this row's payload is binary, not JSON. + var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue(10); // Apply event type allow-list filter (skip events not in the subscription's filter) if (_allowedDotNetTypes != null && dotNetTypeName != null && @@ -133,7 +135,7 @@ SELECT TOP(@batchSize) seq_id, id, stream_id, version, data, type, timestamp, object data; try { - data = _options.Serializer.FromJson(resolvedType, json); + data = _events.DeserializeEventData(resolvedType, json, bdata, _options.Serializer); } catch (Exception ex) { diff --git a/src/Polecat/Events/EventGraph.cs b/src/Polecat/Events/EventGraph.cs index 4361befc..f8d3478c 100644 --- a/src/Polecat/Events/EventGraph.cs +++ b/src/Polecat/Events/EventGraph.cs @@ -440,6 +440,118 @@ public ITagTypeRegistration RegisterTagType(string tableSuffix) where TTag public IReadOnlyList TagTypes => _tagTypes; + // ---- #388: pluggable binary event serialization ------------------------------------------ + // + // Explicit per-type registrations from UseBinarySerializer(...). Types marked with + // [BinaryEvent] but not registered explicitly fall back to DefaultBinarySerializer; that + // resolution is cached in _binarySerializerResolution so the attribute probe (and the throw for a + // misconfigured type) happens once per event type rather than once per row. + private readonly ConcurrentDictionary _binarySerializerByType = new(); + private readonly ConcurrentDictionary _binarySerializerResolution = new(); + + /// + /// Store-wide fallback used for event types marked with + /// that have no explicit per-type registration. Null by + /// default, which leaves every event type on the JSON path. + /// + public IEventBinarySerializer? DefaultBinarySerializer + { + get => _defaultBinarySerializer; + set + { + _defaultBinarySerializer = value; + _binarySerializerResolution.Clear(); + } + } + + private IEventBinarySerializer? _defaultBinarySerializer; + + /// + /// Opt into binary serialization (#388): its payload is written + /// to the bdata column instead of data, and read back through the same + /// serializer. Wins over + . + /// + public EventGraph UseBinarySerializer(IEventBinarySerializer serializer) where TEvent : notnull + { + ArgumentNullException.ThrowIfNull(serializer); + _binarySerializerByType[typeof(TEvent)] = serializer; + _binarySerializerResolution.Clear(); + return this; + } + + /// + /// The governing , or null + /// when that type stays on the JSON path. Explicit registration beats + /// + . + /// + /// + /// The type carries but no serializer is configured for it. + /// Deliberately a throw rather than a silent fall back to JSON: a store that quietly ignored + /// the attribute would have write-amplification characteristics that do not match its + /// configuration, which is the whole reason the feature exists. + /// + internal IEventBinarySerializer? ResolveBinarySerializerFor(Type eventType) + => _binarySerializerResolution.GetOrAdd(eventType, static (type, graph) => + { + if (graph._binarySerializerByType.TryGetValue(type, out var explicitSerializer)) + { + return explicitSerializer; + } + + if (!type.IsDefined(typeof(BinaryEventAttribute), inherit: false)) + { + return null; + } + + return graph._defaultBinarySerializer ?? throw new InvalidOperationException( + $"Event type '{type.FullName}' is marked with [BinaryEvent] but no IEventBinarySerializer " + + $"is registered. Either call opts.Events.UseBinarySerializer<{type.Name}>(...) explicitly, " + + "or set opts.Events.DefaultBinarySerializer to a store-wide fallback."); + }, this); + + /// + /// The bdata bytes for an event on the write path, or null when the event's type stays + /// on the JSON path. The two are mutually exclusive per row — see + /// . + /// + internal byte[]? SerializeEventBdata(IEvent @event) + { + // Deliberately NOT short-circuited on UsesBinaryEventSerialization: a type marked + // [BinaryEvent] in a store with no serializer configured has to throw here, and skipping the + // resolve for "performance" would turn that into a silent write of JSON. The resolve is a + // ConcurrentDictionary hit cached per event type, so a JSON-only store pays one lookup per + // appended event and nothing more. + var eventType = @event.EventType ?? @event.Data.GetType(); + var serializer = ResolveBinarySerializerFor(eventType); + return serializer?.Serialize(eventType, @event.Data); + } + + /// + /// What goes in the data column of a binary event's row. An empty JSON object rather + /// than NULL, because data is NOT NULL and typed json on SQL Server 2025 — a + /// row still has to hold something the engine will parse. + /// + internal const string JsonPlaceholderForBinaryEvent = "{}"; + + /// + /// The per-row read counterpart of : + /// being non-null is the on-row discriminator, so JSON rows written before the feature was + /// switched on keep deserializing through unchanged. + /// + internal object DeserializeEventData(Type resolvedType, string json, byte[]? bdata, ISerializer serializer) + { + if (bdata is null) return serializer.FromJson(resolvedType, json); + + var binary = ResolveBinarySerializerFor(resolvedType) + ?? throw new InvalidOperationException( + $"A pc_events row for '{resolvedType.FullName}' has a non-null bdata column but no " + + "IEventBinarySerializer is registered for that type. Configure it with " + + $"opts.Events.UseBinarySerializer<{resolvedType.Name}>(...) or set " + + "opts.Events.DefaultBinarySerializer — the event cannot be read without it."); + + return binary.Deserialize(resolvedType, bdata); + } + /// /// All currently registered event types. /// diff --git a/src/Polecat/Events/EventOperations.cs b/src/Polecat/Events/EventOperations.cs index 63a14485..07089002 100644 --- a/src/Polecat/Events/EventOperations.cs +++ b/src/Polecat/Events/EventOperations.cs @@ -512,11 +512,17 @@ public void TombstoneStream(string streamKey) public void OverwriteEvent(IEvent @event) { - var serializedData = _sessionBase.Serializer.ToJson(@event.Data); + // #388: same data/bdata split as the append path — a binary event's masked payload has to go + // back through its IEventBinarySerializer, not into the JSON column. + var serializedBdata = _events.SerializeEventBdata(@event); + var serializedData = serializedBdata is null + ? _sessionBase.Serializer.ToJson(@event.Data) + : EventGraph.JsonPlaceholderForBinaryEvent; var serializedHeaders = @event.Headers != null ? _sessionBase.Serializer.ToJson(@event.Headers) : null; - _workTracker.Add(new Protected.OverwriteEventOperation(_events, @event, serializedData, serializedHeaders)); + _workTracker.Add(new Protected.OverwriteEventOperation( + _events, @event, serializedData, serializedBdata, serializedHeaders)); } public Guid CompletelyReplaceEvent(long sequence, T eventBody) where T : class @@ -527,9 +533,14 @@ public Guid CompletelyReplaceEvent(long sequence, T eventBody) where T : clas _events.AddEventType(typeof(T)); var mapping = _events.EventMappingFor(typeof(T)); - var serializedData = _sessionBase.Serializer.ToJson(eventBody); + // #388: the REPLACEMENT body's type decides the row's format, not the replaced row's. + var binary = _events.ResolveBinarySerializerFor(typeof(T)); + var serializedBdata = binary?.Serialize(typeof(T), eventBody); + var serializedData = serializedBdata is null + ? _sessionBase.Serializer.ToJson(eventBody) + : EventGraph.JsonPlaceholderForBinaryEvent; var op = new Protected.ReplaceEventOperation( - _events, sequence, serializedData, mapping.EventTypeName, mapping.DotNetTypeName); + _events, sequence, serializedData, serializedBdata, mapping.EventTypeName, mapping.DotNetTypeName); _workTracker.Add(op); return op.Id; @@ -868,10 +879,10 @@ public async Task> QueryByTagsAsync(EventTagQuery query, var eventOptions = _events.EventOptions; // Build SELECT columns matching the event reader format - var selectColumns = "e.seq_id, e.id, e.stream_id, e.version, e.data, e.type, e.timestamp, e.tenant_id, e.dotnet_type, e.is_archived"; - if (eventOptions.EnableCorrelationId) selectColumns += ", e.correlation_id"; - if (eventOptions.EnableCausationId) selectColumns += ", e.causation_id"; - if (eventOptions.EnableHeaders) selectColumns += ", e.headers"; + // #388: composed by PcEventsRowReader rather than spelled out here — this projection has to + // stay in lockstep with the canonical one (which now carries bdata at ordinal 10), and two + // hand-written copies of the same column list is exactly how that stops being true. + var selectColumns = Internal.PcEventsRowReader.ComposeSelectColumnsWithAlias(eventOptions, "e"); var sb = new StringBuilder(); sb.Append($"SELECT {selectColumns} FROM [{schema}].[pc_events] e"); @@ -949,11 +960,12 @@ public async Task> QueryByTagsAsync(EventTagQuery query, var tenantId = reader.GetString(7); var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8); var isArchived = reader.GetBoolean(9); + var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue(10); // #388 var resolvedType = _events.ResolveEventType(dotNetTypeName); if (resolvedType == null) continue; - var data = _sessionBase.Serializer.FromJson(resolvedType, json); + var data = _events.DeserializeEventData(resolvedType, json, bdata, _sessionBase.Serializer); var mapping = _events.EventMappingFor(resolvedType); var @event = mapping.Wrap(data); @@ -978,7 +990,7 @@ public async Task> QueryByTagsAsync(EventTagQuery query, @event.StreamKey = rawStreamId.ToString(); } - var metaIndex = 10; + var metaIndex = 11; // #388: ordinal 10 is bdata if (eventOptions.EnableCorrelationId) { @event.CorrelationId = reader.IsDBNull(metaIndex) ? null : reader.GetString(metaIndex); @@ -1075,10 +1087,10 @@ internal static void WriteTagQuerySql(ICommandBuilder builder, EventGraph eventG var schema = eventGraph.DatabaseSchemaName; var eventOptions = eventGraph.EventOptions; - var selectColumns = "e.seq_id, e.id, e.stream_id, e.version, e.data, e.type, e.timestamp, e.tenant_id, e.dotnet_type, e.is_archived"; - if (eventOptions.EnableCorrelationId) selectColumns += ", e.correlation_id"; - if (eventOptions.EnableCausationId) selectColumns += ", e.causation_id"; - if (eventOptions.EnableHeaders) selectColumns += ", e.headers"; + // #388: composed by PcEventsRowReader rather than spelled out here — this projection has to + // stay in lockstep with the canonical one (which now carries bdata at ordinal 10), and two + // hand-written copies of the same column list is exactly how that stops being true. + var selectColumns = Internal.PcEventsRowReader.ComposeSelectColumnsWithAlias(eventOptions, "e"); builder.Append($"SELECT {selectColumns} FROM [{schema}].[pc_events] e"); @@ -1148,11 +1160,12 @@ internal static void WriteTagQuerySql(ICommandBuilder builder, EventGraph eventG var tenantId = reader.GetString(7); var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8); var isArchived = reader.GetBoolean(9); + var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue(10); // #388 var resolvedType = eventGraph.ResolveEventType(dotNetTypeName); if (resolvedType == null) return null; - var data = serializer.FromJson(resolvedType, json); + var data = eventGraph.DeserializeEventData(resolvedType, json, bdata, serializer); var mapping = eventGraph.EventMappingFor(resolvedType); var @event = mapping.Wrap(data); @@ -1165,7 +1178,7 @@ internal static void WriteTagQuerySql(ICommandBuilder builder, EventGraph eventG @event.DotNetTypeName = dotNetTypeName!; @event.IsArchived = isArchived; - var metaIndex = 10; + var metaIndex = 11; // #388: ordinal 10 is bdata if (eventOptions.EnableCorrelationId) { @event.CorrelationId = reader.IsDBNull(metaIndex) ? null : reader.GetString(metaIndex); diff --git a/src/Polecat/Events/IEventBinarySerializer.cs b/src/Polecat/Events/IEventBinarySerializer.cs new file mode 100644 index 00000000..09565aab --- /dev/null +++ b/src/Polecat/Events/IEventBinarySerializer.cs @@ -0,0 +1,42 @@ +namespace Polecat.Events; + +/// +/// Pluggable binary serializer for event data — Polecat's counterpart of Marten's +/// IEventBinarySerializer (, +/// shipped in Marten 9.20.2), at parity so a store-agnostic consumer can wire either flavor. +/// Tracked as polecat#388. +/// +/// +/// +/// Binary serialization is enabled per event type, not store-wide. A store +/// can have JSON events and binary events mixed in the same pc_events table; a row's +/// format is determined by whether its bdata column is NULL (JSON) or not +/// (binary). That is what makes the feature safe to switch on for an existing store with no +/// migration of existing event data — and just as safe to switch back off. +/// +/// +/// Opt in by marking an event type with (resolved against +/// the store-wide opts.Events.DefaultBinarySerializer) or by registering it explicitly +/// with opts.Events.UseBinarySerializer<TEvent>(serializer). An explicit per-type +/// registration wins over the attribute. +/// +/// +/// Implementations must be thread-safe: one instance serves every session in the store. +/// +/// +public interface IEventBinarySerializer +{ + /// + /// Serialize an event data instance to bytes. + /// + /// The runtime CLR type of the event data. + /// The event data to serialize. + byte[] Serialize(Type type, object data); + + /// + /// Deserialize bytes back into an event data instance. + /// + /// The target CLR type to deserialize into. + /// The bytes previously produced by . + object Deserialize(Type type, byte[] data); +} diff --git a/src/Polecat/Events/Internal/PcEventsRowReader.cs b/src/Polecat/Events/Internal/PcEventsRowReader.cs index b14d5fbd..c1649784 100644 --- a/src/Polecat/Events/Internal/PcEventsRowReader.cs +++ b/src/Polecat/Events/Internal/PcEventsRowReader.cs @@ -61,10 +61,16 @@ namespace Polecat.Events.Internal; internal static class PcEventsRowReader { /// - /// Mandatory columns, always projected. Ordinals 0–9. + /// Mandatory columns, always projected. Ordinals 0–10. /// + /// + /// #388 pins bdata at ordinal 10, after the previously-locked 0–9 block and before the + /// optional metadata columns, so the per-row JSON-vs-binary dispatch reads a stable ordinal + /// without disturbing any of the existing ones. The optional metadata slots shift from 10 to 11 + /// — see . + /// internal const string CoreSelectColumns = - "seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived"; + "seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived, bdata"; /// /// Compose plus any optional metadata @@ -93,7 +99,7 @@ internal static string ComposeSelectColumns(EventStoreOptions options) internal static string ComposeSelectColumnsWithAlias(EventStoreOptions options, string alias) { var sb = new StringBuilder( - $"{alias}.seq_id, {alias}.id, {alias}.stream_id, {alias}.version, {alias}.data, {alias}.type, {alias}.timestamp, {alias}.tenant_id, {alias}.dotnet_type, {alias}.is_archived"); + $"{alias}.seq_id, {alias}.id, {alias}.stream_id, {alias}.version, {alias}.data, {alias}.type, {alias}.timestamp, {alias}.tenant_id, {alias}.dotnet_type, {alias}.is_archived, {alias}.bdata"); if (options.EnableCorrelationId) sb.Append($", {alias}.correlation_id"); if (options.EnableCausationId) sb.Append($", {alias}.causation_id"); if (options.EnableHeaders) sb.Append($", {alias}.headers"); @@ -162,6 +168,10 @@ internal static string ComposeSelectColumnsWithAlias(EventStoreOptions options, var tenantId = reader.IsDBNull(7) ? ctx.DefaultTenantId : reader.GetString(7); var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8); var isArchived = reader.GetBoolean(9); + // #388: non-null bdata is the per-row discriminator. Reading it here (rather than inside the + // optional-metadata slot loop) keeps the dispatch on a fixed ordinal and off the JSON path's + // hot loop — a store with no binary events pays one IsDBNull per row. + var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue(10); var resolvedType = ctx.EventGraph.ResolveEventType(dotNetTypeName); if (resolvedType == null) return null; @@ -171,7 +181,7 @@ internal static string ComposeSelectColumnsWithAlias(EventStoreOptions options, // this collapses N dictionary lookups into 1 per distinct type. var mapping = cache.LookupOrAdd(ctx.EventGraph, resolvedType); - var data = ctx.Serializer.FromJson(resolvedType, json); + var data = ctx.EventGraph.DeserializeEventData(resolvedType, json, bdata, ctx.Serializer); var @event = mapping.Wrap(data); @event.Id = eventId; @@ -253,6 +263,10 @@ internal static EventRecord ReadEventRecord( metadata = headerDoc.RootElement.Clone(); } + // #388: a binary event's `data` column holds only the '{}' placeholder, so the explorer's + // JsonElement view of a binary row is that empty object. The explorer is a diagnostic surface + // that deliberately does not deserialize (no ISerializer, no IEventBinarySerializer), so it + // reports what is in the JSON column rather than pretending to decode the bytes. using var doc = JsonDocument.Parse(rawData); var data = doc.RootElement.Clone(); @@ -294,7 +308,8 @@ internal readonly record struct MetadataSlots(int CorrelationIdx, int CausationI public static MetadataSlots Compute(EventStoreOptions options) { - var ordinal = 10; + // #388: ordinal 10 is bdata; the optional metadata block starts at 11. + var ordinal = 11; var correlation = options.EnableCorrelationId ? ordinal++ : Disabled; var causation = options.EnableCausationId ? ordinal++ : Disabled; var headers = options.EnableHeaders ? ordinal++ : Disabled; diff --git a/src/Polecat/Events/Linq/EventLinqQueryProvider.cs b/src/Polecat/Events/Linq/EventLinqQueryProvider.cs index b68eb90c..7c9cc101 100644 --- a/src/Polecat/Events/Linq/EventLinqQueryProvider.cs +++ b/src/Polecat/Events/Linq/EventLinqQueryProvider.cs @@ -143,7 +143,7 @@ or SingleValueMode.Any or SingleValueMode.Sum or SingleValueMode.Min // Full IEvent result set. #256: append the opt-in metadata columns (when enabled) so // hydrated events carry correlation/causation/user_name — EventListHandler reads them // at the trailing ordinals in this same enable order. - var columns = "seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived"; + var columns = Internal.PcEventsRowReader.CoreSelectColumns; var eventOptions = _events.EventOptions; if (eventOptions.EnableCorrelationId) columns += ", correlation_id"; if (eventOptions.EnableCausationId) columns += ", causation_id"; diff --git a/src/Polecat/Events/Linq/EventListHandler.cs b/src/Polecat/Events/Linq/EventListHandler.cs index f67676f6..faf68804 100644 --- a/src/Polecat/Events/Linq/EventListHandler.cs +++ b/src/Polecat/Events/Linq/EventListHandler.cs @@ -8,7 +8,8 @@ namespace Polecat.Events.Linq; /// /// Reads IEvent objects from a multi-column result set on pc_events. -/// Column layout: seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived +/// Column layout: seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, +/// is_archived, bdata (#388), then the optional metadata columns. /// [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Class-level: hydrates IEvent via EventGraph.Wrap (which routes through ISerializer.FromJson). Event types are preserved by EventGraph registration on the caller side per the AOT publishing guide.")] @@ -42,11 +43,13 @@ public async Task> HandleAsync(DbDataReader reader, Cancel var tenantId = sqlReader.GetString(7); var dotNetTypeName = sqlReader.IsDBNull(8) ? null : sqlReader.GetString(8); var isArchived = sqlReader.GetBoolean(9); + // #388: non-null bdata means the payload is binary, not in the JSON `data` column. + var bdata = sqlReader.IsDBNull(10) ? null : sqlReader.GetFieldValue(10); var resolvedType = _events.ResolveEventType(dotNetTypeName); if (resolvedType == null) continue; - var data = _serializer.FromJson(resolvedType, json); + var data = _events.DeserializeEventData(resolvedType, json, bdata, _serializer); var mapping = _events.EventMappingFor(resolvedType); var @event = mapping.Wrap(data); @@ -71,7 +74,7 @@ public async Task> HandleAsync(DbDataReader reader, Cancel // #256: read the opt-in metadata columns appended to the full-event SELECT (when enabled), // in the same enable order the query provider emitted them. var options = _events.EventOptions; - var ordinal = 10; + var ordinal = 11; // #388: ordinal 10 is bdata if (options.EnableCorrelationId) { @event.CorrelationId = sqlReader.IsDBNull(ordinal) ? null : sqlReader.GetString(ordinal); diff --git a/src/Polecat/Events/Protected/OverwriteEventOperation.cs b/src/Polecat/Events/Protected/OverwriteEventOperation.cs index 7018b7c2..3b5046e7 100644 --- a/src/Polecat/Events/Protected/OverwriteEventOperation.cs +++ b/src/Polecat/Events/Protected/OverwriteEventOperation.cs @@ -11,13 +11,16 @@ internal class OverwriteEventOperation : Polecat.Internal.IStorageOperation private readonly EventGraph _events; private readonly IEvent _event; private readonly string _serializedData; + private readonly byte[]? _serializedBdata; private readonly string? _serializedHeaders; - public OverwriteEventOperation(EventGraph events, IEvent @event, string serializedData, string? serializedHeaders) + public OverwriteEventOperation(EventGraph events, IEvent @event, string serializedData, + byte[]? serializedBdata, string? serializedHeaders) { _events = events; _event = @event; _serializedData = serializedData; + _serializedBdata = serializedBdata; _serializedHeaders = serializedHeaders; } @@ -26,24 +29,32 @@ public OverwriteEventOperation(EventGraph events, IEvent @event, string serializ public void ConfigureCommand(ICommandBuilder builder) { - if (_serializedHeaders != null) + builder.Append($"UPDATE {_events.EventsTableName} SET data = "); + builder.AppendParameter(_serializedData); + + // #388: a binary event's payload lives in bdata, so masking has to rewrite THAT — rewriting + // only `data` would leave the original, unmasked payload readable in bdata, which for a GDPR + // masking operation is the whole point missed. The row keeps its own format either way: + // binary events stay binary (data holds the '{}' placeholder), JSON events stay JSON. + builder.Append(", bdata = "); + if (_serializedBdata is null) { - builder.Append($"UPDATE {_events.EventsTableName} SET data = "); - builder.AppendParameter(_serializedData); - builder.Append(", headers = "); - builder.AppendParameter(_serializedHeaders); - builder.Append(" WHERE seq_id = "); - builder.AppendParameter(_event.Sequence); - builder.Append(";"); + builder.Append("NULL"); } else { - builder.Append($"UPDATE {_events.EventsTableName} SET data = "); - builder.AppendParameter(_serializedData); - builder.Append(" WHERE seq_id = "); - builder.AppendParameter(_event.Sequence); - builder.Append(";"); + builder.AppendParameter(_serializedBdata); } + + if (_serializedHeaders != null) + { + builder.Append(", headers = "); + builder.AppendParameter(_serializedHeaders); + } + + builder.Append(" WHERE seq_id = "); + builder.AppendParameter(_event.Sequence); + builder.Append(";"); } public Task PostprocessAsync(DbDataReader reader, IList exceptions, CancellationToken token) diff --git a/src/Polecat/Events/Protected/ReplaceEventOperation.cs b/src/Polecat/Events/Protected/ReplaceEventOperation.cs index 7d06f128..d6f5cdf2 100644 --- a/src/Polecat/Events/Protected/ReplaceEventOperation.cs +++ b/src/Polecat/Events/Protected/ReplaceEventOperation.cs @@ -11,16 +11,18 @@ internal class ReplaceEventOperation : Polecat.Internal.IStorageOperation private readonly EventGraph _events; private readonly long _sequence; private readonly string _serializedData; + private readonly byte[]? _serializedBdata; private readonly string _eventTypeName; private readonly string _dotNetTypeName; private readonly Guid _newId; public ReplaceEventOperation(EventGraph events, long sequence, string serializedData, - string eventTypeName, string dotNetTypeName) + byte[]? serializedBdata, string eventTypeName, string dotNetTypeName) { _events = events; _sequence = sequence; _serializedData = serializedData; + _serializedBdata = serializedBdata; _eventTypeName = eventTypeName; _dotNetTypeName = dotNetTypeName; _newId = Guid.NewGuid(); @@ -34,6 +36,20 @@ public void ConfigureCommand(ICommandBuilder builder) { builder.Append($"UPDATE {_events.EventsTableName} SET data = "); builder.AppendParameter(_serializedData); + + // #388: the replacement body's own event type decides the row's format, so bdata is set (or + // cleared) in the same statement — a replacement that switched a row from binary to JSON + // without clearing bdata would keep reading the OLD payload. + builder.Append(", bdata = "); + if (_serializedBdata is null) + { + builder.Append("NULL"); + } + else + { + builder.AppendParameter(_serializedBdata); + } + builder.Append(", timestamp = SYSDATETIMEOFFSET(), type = "); builder.AppendParameter(_eventTypeName); builder.Append(", dotnet_type = "); diff --git a/src/Polecat/Events/Protected/StreamCompacting.cs b/src/Polecat/Events/Protected/StreamCompacting.cs index 37096457..cb8cbaba 100644 --- a/src/Polecat/Events/Protected/StreamCompacting.cs +++ b/src/Polecat/Events/Protected/StreamCompacting.cs @@ -70,11 +70,21 @@ await archiver.MaybeArchiveAsync(session, request, events, request.CancellationT var compacted = new Compacted(aggregate!, request.StreamId ?? Guid.Empty, request.StreamKey ?? string.Empty); - var serializedData = session.Serializer.ToJson(compacted); - var mapping = session.Options.EventGraph.EventMappingFor(typeof(Compacted)); + var graph = session.Options.EventGraph; + + // #388: the Compacted snapshot follows the same data/bdata rule as any other event — + // binary if Compacted itself is opted in, JSON otherwise. The event being REPLACED may + // have been the other format; ReplaceEventOperation writes both columns so the row ends up + // consistent either way. + var binary = graph.ResolveBinarySerializerFor(typeof(Compacted)); + var serializedBdata = binary?.Serialize(typeof(Compacted), compacted); + var serializedData = serializedBdata is null + ? session.Serializer.ToJson(compacted) + : EventGraph.JsonPlaceholderForBinaryEvent; + var mapping = graph.EventMappingFor(typeof(Compacted)); var replaceOp = new ReplaceEventOperation( - session.Options.EventGraph, request.Sequence, serializedData, + graph, request.Sequence, serializedData, serializedBdata, mapping.EventTypeName, mapping.DotNetTypeName); session.WorkTracker.Add(replaceOp); diff --git a/src/Polecat/Events/Schema/EventsTable.cs b/src/Polecat/Events/Schema/EventsTable.cs index 5bd47463..c02b69a8 100644 --- a/src/Polecat/Events/Schema/EventsTable.cs +++ b/src/Polecat/Events/Schema/EventsTable.cs @@ -44,6 +44,17 @@ public EventsTable(EventGraph events) // Event data — SQL Server 2025 native JSON type by default AddColumn("data", events.JsonColumnType).NotNull(); + // #388: binary event payload, the SQL Server counterpart of Marten's bytea `bdata` + // (marten#4515). Nullable and unconditional on purpose: `bdata IS NULL` is the per-ROW + // discriminator between a JSON event and a binary one, so JSON and binary events coexist in + // this table per event type. That is what makes the feature switchable on an existing store + // with no migration of existing event data — rows written before the column existed simply + // have bdata = NULL and keep reading through the JSON path — and just as safely switchable + // back off. Adding it always (rather than only when a serializer is configured) keeps the + // read projection one fixed shape; Weasel's additive delta adds the column on the next + // schema apply. + AddColumn("bdata", "varbinary(max)").AllowNulls(); + // Event type name for deserialization AddColumn("type", "varchar(500)").NotNull(); diff --git a/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs b/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs index eb2ecbb0..5500baa6 100644 --- a/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs +++ b/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs @@ -135,7 +135,7 @@ public void ConfigureCommand(Weasel.Core.ICommandBuilder builder, IStorageSessio builder.Append(", "); } - builder.Append("id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type"); + builder.Append("id, stream_id, version, data, bdata, type, timestamp, tenant_id, dotnet_type"); if (options.EnableCorrelationId) builder.Append(", correlation_id"); if (options.EnableCausationId) builder.Append(", causation_id"); if (options.EnableHeaders) builder.Append(", headers"); @@ -164,8 +164,18 @@ public void ConfigureCommand(Weasel.Core.ICommandBuilder builder, IStorageSessio builder.Append(", "); Bind(builder, @event.Version, StorageColumnType.Long); + // #388: a binary event writes the '{}' placeholder to `data` and its payload to `bdata`; + // a JSON event writes the payload to `data` and NULL to `bdata`. Exactly one of the two + // is populated per row, and `bdata IS NULL` is what the read path dispatches on. + var bdata = _graph.SerializeEventBdata(@event); + + builder.Append(", "); + Bind(builder, + bdata is null ? session.Serializer.ToJson(@event.Data) : EventGraph.JsonPlaceholderForBinaryEvent, + StorageColumnType.Json); + builder.Append(", "); - Bind(builder, session.Serializer.ToJson(@event.Data), StorageColumnType.Json); + Bind(builder, (object?)bdata ?? DBNull.Value, StorageColumnType.Binary); builder.Append(", "); Bind(builder, @event.EventTypeName, StorageColumnType.String); diff --git a/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs b/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs index 21d1de1d..4fce484e 100644 --- a/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs +++ b/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs @@ -50,11 +50,14 @@ public QuickEventStorageDescriptor BuildQuickDescriptor(EventRegistry registry, quickAppendEventsSql: $"insert into {graph.EventsTableName} ", insertStreamSql: $"insert into {graph.StreamsTableName} (...) values (...)", updateStreamVersionSql: $"update {graph.StreamsTableName} set version = ... where ...", - // Polecat is STJ/JSON-only — no binary event serialization. Data is always JSON; bdata is - // always null. Serialize through the session serializer at write time (see the append op); - // the descriptor closure below covers any shared-op path that reaches for it. - serializeEventData: e => serializer.ToJson(e.Data), - serializeEventBdata: _ => null) + // #388: event data is JSON unless the event's type is opted into binary serialization, in + // which case `data` carries the '{}' placeholder and the payload goes to `bdata`. Exactly + // one of the two closures produces a payload for any given event. These cover any shared-op + // path that reaches for them; the SQL Server append operation binds both directly. + serializeEventData: e => graph.SerializeEventBdata(e) is null + ? serializer.ToJson(e.Data) + : EventGraph.JsonPlaceholderForBinaryEvent, + serializeEventBdata: graph.SerializeEventBdata) { IsGuidStreamIdentity = isGuid, Dialect = dialect, diff --git a/src/Polecat/Storage/SqlServerStorageDialect.cs b/src/Polecat/Storage/SqlServerStorageDialect.cs index 7284f75c..949a74d8 100644 --- a/src/Polecat/Storage/SqlServerStorageDialect.cs +++ b/src/Polecat/Storage/SqlServerStorageDialect.cs @@ -95,6 +95,8 @@ public void SetParameterType(System.Data.Common.DbParameter parameter, StorageCo // Polecat binds JSON as string values throughout; SQL Server 2025's native JSON // column type (and nvarchar-backed JSON) accepts nvarchar input. StorageColumnType.Json => SqlDbType.NVarChar, + // #388: the binary event payload column (pc_events.bdata) is varbinary(max). + StorageColumnType.Binary => SqlDbType.VarBinary, _ => throw new ArgumentOutOfRangeException(nameof(type)) }; diff --git a/src/Polecat/StoreOptions.cs b/src/Polecat/StoreOptions.cs index d78d31c0..5ecb44c9 100644 --- a/src/Polecat/StoreOptions.cs +++ b/src/Polecat/StoreOptions.cs @@ -490,6 +490,32 @@ public void AddEventTypes(IEnumerable eventTypes) foreach (var eventType in eventTypes) EventGraph!.AddEventType(eventType); } + /// + /// #388: store-wide fallback for event types + /// marked with that have no explicit per-type + /// registration via . Null by default — every event type + /// stays on the JSON path. Mirrors Marten's opts.Events.DefaultBinarySerializer. + /// + public Polecat.Events.IEventBinarySerializer? DefaultBinarySerializer + { + get => EventGraph!.DefaultBinarySerializer; + set => EventGraph!.DefaultBinarySerializer = value; + } + + /// + /// #388: opt into binary serialization — its payload is written to + /// the pc_events.bdata column instead of data, and read back through the same + /// serializer. Per event type rather than store-wide, so JSON and binary rows coexist in one + /// table and the feature can be switched on (or back off) for an existing store with no data + /// migration. Mirrors Marten's opts.Events.UseBinarySerializer<TEvent>(serializer). + /// + public EventStoreOptions UseBinarySerializer(Polecat.Events.IEventBinarySerializer serializer) + where TEvent : notnull + { + EventGraph!.UseBinarySerializer(serializer); + return this; + } + /// /// Register a tag type for Dynamic Consistency Boundary (DCB) support. /// Creates a tag table with an auto-generated suffix.