Skip to content

Move the tag rows when archiving a stream under archived-stream partitioning #5679

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jeremydmiller merged 1 commit into from
Oct 15, 2026
Merged

Move the tag rows when archiving a stream under archived-stream partitioning #5679

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
283 changes: 283 additions & 0 deletions src/EventSourcingTests/Archiving/Bug_5675_archiving_a_tagged_stream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,283 @@
using System;
using System.Threading.Tasks;
using JasperFx.Events;
using JasperFx.Events.Tags;
using Marten;
using Marten.Events;
using Marten.Storage;
using Marten.Testing.Harness;
using Npgsql;
using Shouldly;
using Weasel.Core;
using Weasel.Postgresql;
using Xunit;

namespace EventSourcingTests.Archiving;

/// <summary>
/// #5675. With <c>UseArchivedStreamPartitioning</c> on, each <c>mt_event_tag_*</c> table is itself
/// partitioned by <c>is_archived</c> and carries a composite foreign key <c>(seq_id, is_archived)</c> to
/// <c>mt_events</c>. <c>mt_archive_stream</c> copied the events into the archived partition and deleted
/// the originals without moving the tag rows, so the delete was refused: archiving any tagged stream
/// came back as a foreign key violation.
/// </summary>
public class Bug_5675_archiving_a_tagged_stream: OneOffConfigurationsContext
{
public record LedgerOpened(string Name);

public record LedgerPosted(decimal Amount);

public record LedgerId(Guid Value);

public record AuditorId(string Value);

private void configure(TenancyStyle tenancy = TenancyStyle.Single, DcbStorageMode mode = DcbStorageMode.TagTables,
bool twoTagTypes = false)
{
StoreOptions(opts =>
{
opts.Events.UseArchivedStreamPartitioning = true;
opts.Events.TenancyStyle = tenancy;
opts.Events.DcbStorageMode = mode;
opts.Events.AddEventType<LedgerOpened>();
opts.Events.AddEventType<LedgerPosted>();

opts.Events.RegisterTagType<LedgerId, Guid>(id => id.Value, "ledger");

// A second tag table, because mt_archive_stream has to emit statements for every registered
// tag type -- one tag type passing proves nothing about the loop.
if (twoTagTypes) opts.Events.RegisterTagType<AuditorId, string>(id => id.Value, "auditor");
});
}

[Fact]
public async Task archiving_a_tagged_stream_moves_its_tag_rows()
{
configure();

var archived = Guid.NewGuid();
var kept = Guid.NewGuid();
var archivedTag = new LedgerId(Guid.NewGuid());
var keptTag = new LedgerId(Guid.NewGuid());

await using (var session = theStore.LightweightSession())
{
var opened = session.Events.BuildEvent(new LedgerOpened("archived"));
opened.WithTag(archivedTag);
var posted = session.Events.BuildEvent(new LedgerPosted(10));
posted.WithTag(archivedTag);
session.Events.StartStream(archived, opened, posted);

var neighbour = session.Events.BuildEvent(new LedgerOpened("kept"));
neighbour.WithTag(keptTag);
session.Events.StartStream(kept, neighbour);

await session.SaveChangesAsync();
}

await using (var session = theStore.LightweightSession())
{
session.Events.ArchiveStream(archived);
await session.SaveChangesAsync();
}

// The two tag rows moved into the archived partition with their events, and none was left behind
// in the live one
(await countTagRows("ledger", archivedTag.Value, isArchived: true)).ShouldBe(2);
(await countTagRows("ledger", archivedTag.Value, isArchived: false)).ShouldBe(0);

// The neighbouring stream's tag is untouched
(await countTagRows("ledger", keptTag.Value, isArchived: false)).ShouldBe(1);
(await countTagRows("ledger", keptTag.Value, isArchived: true)).ShouldBe(0);
}

[Fact]
public async Task archiving_moves_every_registered_tag_table()
{
configure(twoTagTypes: true);

var streamId = Guid.NewGuid();
var ledger = new LedgerId(Guid.NewGuid());
var auditor = new AuditorId("auditor-" + Guid.NewGuid().ToString("N"));

await using (var session = theStore.LightweightSession())
{
var opened = session.Events.BuildEvent(new LedgerOpened("both"));
opened.WithTag(ledger, auditor);
session.Events.StartStream(streamId, opened);
await session.SaveChangesAsync();
}

await using (var session = theStore.LightweightSession())
{
session.Events.ArchiveStream(streamId);
await session.SaveChangesAsync();
}

(await countTagRows("ledger", ledger.Value, isArchived: true)).ShouldBe(1);
(await countTagRows("ledger", ledger.Value, isArchived: false)).ShouldBe(0);
(await countTagRows("auditor", auditor.Value, isArchived: true)).ShouldBe(1);
(await countTagRows("auditor", auditor.Value, isArchived: false)).ShouldBe(0);
}

[Fact]
public async Task archiving_a_tagged_stream_under_conjoined_tenancy_leaves_other_tenants_alone()
{
configure(TenancyStyle.Conjoined);

var streamId = Guid.NewGuid();
var tag = new LedgerId(Guid.NewGuid());

// The same stream id and the same tag value in two tenants: the tenant predicate in the archive
// function has to tell them apart, and the subquery's tenant_id must not be read as the tag
// table's own.
foreach (var tenant in new[] { "one", "two" })
{
await using var session = theStore.LightweightSession(tenant);
var opened = session.Events.BuildEvent(new LedgerOpened(tenant));
opened.WithTag(tag);
session.Events.StartStream(streamId, opened);
await session.SaveChangesAsync();
}

await using (var session = theStore.LightweightSession("one"))
{
session.Events.ArchiveStream(streamId);
await session.SaveChangesAsync();
}

(await countTagRows("ledger", tag.Value, isArchived: true, tenantId: "one")).ShouldBe(1);
(await countTagRows("ledger", tag.Value, isArchived: false, tenantId: "one")).ShouldBe(0);

(await countTagRows("ledger", tag.Value, isArchived: false, tenantId: "two")).ShouldBe(1);
(await countTagRows("ledger", tag.Value, isArchived: true, tenantId: "two")).ShouldBe(0);
}

[Fact]
public async Task an_untagged_stream_beside_a_registered_tag_type_still_archives()
{
// The tag statements run whether or not this particular stream has tag rows; nothing may assume
// there is at least one.
configure();

var streamId = Guid.NewGuid();

await using (var session = theStore.LightweightSession())
{
session.Events.StartStream(streamId, new LedgerOpened("untagged"));
await session.SaveChangesAsync();
}

await using (var session = theStore.LightweightSession())
{
session.Events.ArchiveStream(streamId);
await session.SaveChangesAsync();
}

(await countEventRows(streamId, isArchived: true)).ShouldBe(1);
(await countEventRows(streamId, isArchived: false)).ShouldBe(0);
}

[Fact]
public async Task archiving_a_tagged_stream_in_hstore_mode_still_works()
{
// HStore mode has no tag tables at all -- the tags ride along on mt_events.tags, and so move with
// the row. The archive function must emit no tag statements for a table that does not exist.
configure(mode: DcbStorageMode.HStore);

var streamId = Guid.NewGuid();
var tag = new LedgerId(Guid.NewGuid());

await using (var session = theStore.LightweightSession())
{
var opened = session.Events.BuildEvent(new LedgerOpened("hstore"));
opened.WithTag(tag);
session.Events.StartStream(streamId, opened);
await session.SaveChangesAsync();
}

await using (var session = theStore.LightweightSession())
{
session.Events.ArchiveStream(streamId);
await session.SaveChangesAsync();
}

(await countEventRows(streamId, isArchived: true)).ShouldBe(1);
(await countEventRows(streamId, isArchived: false)).ShouldBe(0);
}

[Fact]
public async Task a_tag_query_still_sees_the_archived_events_which_is_a_separate_bug()
{
// Pinning what it DOES, not what it should. #5675 asked that tag queries "behave as documented"
// after archiving, and they do not behave like any other event query: QueryAllRawEvents() excludes
// archived events by default, and QueryByTagsAsync / EventsExistAsync do not. Verified independent
// of this fix -- the same two counts come back with UseArchivedStreamPartitioning OFF, which the
// archive function change here never touches -- so it is pre-existing and filed as #5677. Change
// this test when that one is fixed; it is here so the current answer is written down rather than
// discovered again.
configure();

var archived = Guid.NewGuid();
var kept = Guid.NewGuid();
var tag = new LedgerId(Guid.NewGuid());

await using (var session = theStore.LightweightSession())
{
var first = session.Events.BuildEvent(new LedgerOpened("archived"));
first.WithTag(tag);
session.Events.StartStream(archived, first);

var second = session.Events.BuildEvent(new LedgerOpened("kept"));
second.WithTag(tag);
session.Events.StartStream(kept, second);

await session.SaveChangesAsync();
}

await using (var session = theStore.LightweightSession())
{
session.Events.ArchiveStream(archived);
await session.SaveChangesAsync();
}

await using (var query = theStore.LightweightSession())
{
// The archived row moved partitions and is still reachable by tag -- which is the point worth
// pinning here: the move does not LOSE the tag rows either
(await query.Events.QueryByTagsAsync(new EventTagQuery().Or<LedgerId>(tag))).Count.ShouldBe(2);

// While the ordinary event query does exclude it
(await query.Events.QueryAllRawEvents().ToListAsync()).Count.ShouldBe(1);
}
}

private async Task<long> countTagRows(string suffix, object value, bool isArchived, string? tenantId = null)
{
await using var connection = new NpgsqlConnection(ConnectionSource.ConnectionString);
await connection.OpenAsync();

var sql =
$"select count(*) from {SchemaName}.mt_event_tag_{suffix} where value = :value and is_archived = :archived";
if (tenantId != null) sql += " and tenant_id = :tenant";

await using var command = connection.CreateCommand(sql)
.With("value", value)
.With("archived", isArchived);
if (tenantId != null) command.With("tenant", tenantId);

return (long)(await command.ExecuteScalarAsync())!;
}

private async Task<long> countEventRows(Guid streamId, bool isArchived)
{
await using var connection = new NpgsqlConnection(ConnectionSource.ConnectionString);
await connection.OpenAsync();
await using var command = connection
.CreateCommand(
$"select count(*) from {SchemaName}.mt_events where stream_id = :stream and is_archived = :archived")
.With("stream", streamId)
.With("archived", isArchived);
return (long)(await command.ExecuteScalarAsync())!;
}
}
69 changes: 68 additions & 1 deletion src/Marten/Events/Archiving/ArchiveStreamFunction.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,20 +80,87 @@ private void writeWithPartitioning(TextWriter writer, string argList, string ten

// Serialize archives of the same active stream before copying it. Filtering alone lets
// concurrent calls select the same row and collide in the archived partition.
//
// #5675: the tag rows move with their events, and the ORDER of these statements is the whole fix.
// Each mt_event_tag_* table is itself partitioned by is_archived and carries a composite foreign
// key (seq_id, is_archived) to mt_events, which is NOT deferrable -- so every statement here has to
// leave both sides of that key consistent on its own:
//
// 1. the archived event rows are inserted FIRST, so (seq_id, TRUE) exists to point at;
// 2. then the archived tag rows, which reference it;
// 3. then the original tag rows are deleted, BEFORE the events they still reference;
// 4. and only then the original event rows.
//
// Before this, step 4 ran with the tag rows still pointing at (seq_id, FALSE), so archiving a
// tagged stream came back as a foreign key violation on the delete.
var archiveTags = tagMoves(archived: true);
var deleteTags = tagMoves(archived: false);

writer.WriteLine($@"
CREATE OR REPLACE FUNCTION {_events.DatabaseSchemaName}.{Name}({argList}) RETURNS VOID LANGUAGE plpgsql AS
$function$
BEGIN
perform 1 from {_events.DatabaseSchemaName}.mt_streams where id = streamid and {IsArchivedColumn.ColumnName} = FALSE {tenantWhere} for update;
insert into {_events.DatabaseSchemaName}.mt_streams ({streamInsertColumns}) select {streamColumnList}, TRUE from {_events.DatabaseSchemaName}.mt_streams where id = streamid and {IsArchivedColumn.ColumnName} = FALSE {tenantWhere};
insert into {_events.DatabaseSchemaName}.mt_events ({eventInsertColumns}) select {eventColumnList}, TRUE from {_events.DatabaseSchemaName}.mt_events where stream_id = streamid and {IsArchivedColumn.ColumnName} = FALSE {tenantWhere};
insert into {_events.DatabaseSchemaName}.mt_events ({eventInsertColumns}) select {eventColumnList}, TRUE from {_events.DatabaseSchemaName}.mt_events where stream_id = streamid and {IsArchivedColumn.ColumnName} = FALSE {tenantWhere};{archiveTags}{deleteTags}
delete from {_events.DatabaseSchemaName}.mt_events where stream_id = streamid and {IsArchivedColumn.ColumnName} = FALSE {tenantWhere};
delete from {_events.DatabaseSchemaName}.mt_streams where id = streamid and {IsArchivedColumn.ColumnName} = FALSE {tenantWhere};
END;
$function$;
");
}

/// <summary>
/// The statements that copy each tag table's rows into its archived partition
/// (<paramref name="archived" /> true) or remove the originals (false). Empty when there are no tag
/// tables to move — no registered tag types, or HStore mode, where tags live inline on
/// <c>mt_events.tags</c> and travel with the row.
/// </summary>
/// <remarks>
/// The column list is explicit for the same reason #4619 made the event and stream inserts explicit:
/// an ALTER TABLE ADD COLUMN on an existing tag table appends the column physically, and a
/// positional insert would then silently load the wrong column.
/// </remarks>
private string tagMoves(bool archived)
{
if (_events.DcbStorageMode == DcbStorageMode.HStore) return "";

var schema = _events.DatabaseSchemaName;
var isConjoined = _events.TenancyStyle == TenancyStyle.Conjoined;
var writer = new StringWriter();

foreach (var registration in _events.TagTypes)
{
var table = $"{schema}.mt_event_tag_{registration.TableSuffix}";

// Aliased because both the tag table and mt_events carry tenant_id under conjoined tenancy,
// and the subquery would otherwise resolve the outer predicate's column to the inner table.
var tagTenantWhere = isConjoined ? " and t.tenant_id = tenantid" : "";
var ofThisStream =
$"t.seq_id in (select e.seq_id from {schema}.mt_events e where e.stream_id = streamid and e.{IsArchivedColumn.ColumnName} = FALSE{(isConjoined ? " and e.tenant_id = tenantid" : "")})";

if (archived)
{
var columns = isConjoined
? $"value, tenant_id, seq_id, {IsArchivedColumn.ColumnName}"
: $"value, seq_id, {IsArchivedColumn.ColumnName}";
var source = isConjoined ? "t.value, t.tenant_id, t.seq_id" : "t.value, t.seq_id";

writer.WriteLine();
writer.Write(
$" insert into {table} ({columns}) select {source}, TRUE from {table} t where {ofThisStream} and t.{IsArchivedColumn.ColumnName} = FALSE{tagTenantWhere};");
}
else
{
writer.WriteLine();
writer.Write(
$" delete from {table} t where {ofThisStream} and t.{IsArchivedColumn.ColumnName} = FALSE{tagTenantWhere};");
}
}

return writer.ToString();
}

private void writeSimple(TextWriter writer, string argList, string tenantWhere)
{
writer.WriteLine($@"
Expand Down
Loading