From 15aa51c478fc25f98d37e3dd94b902c99063f9a6 Mon Sep 17 00:00:00 2001 From: Will Baker Date: Tue, 7 Jul 2026 22:05:40 +0000 Subject: [PATCH] message: add CONTROL flag for out-of-band metadata messages Flow's backfill-truncation feature publishes control messages into a collection's journals to mark where a re-backfill supersedes prior data. They carry no content of their own, but need a flag distinct from the transaction-semantics bits so consumers can recognize and drop them. Add Flag_CONTROL at bit 2, orthogonal to the low two transaction bits. The Sequencer masks the bit off when classifying a message and sequences it by its underlying Flag_OUTSIDE_TXN / Flag_CONTINUE_TXN / Flag_ACK_TXN, rather than tripping the unknown-flags error path; the flag is retained on the delivered envelope for the consumer to act on. --- message/sequencer.go | 7 ++++++- message/sequencer_test.go | 31 +++++++++++++++++++++++++++++++ message/uuid.go | 21 +++++++++++++++++---- message/uuid_test.go | 13 +++++++++++++ 4 files changed, 67 insertions(+), 5 deletions(-) diff --git a/message/sequencer.go b/message/sequencer.go index 2b58c504..40d01578 100644 --- a/message/sequencer.go +++ b/message/sequencer.go @@ -201,8 +201,13 @@ func (w *Sequencer) QueueUncommitted(env Envelope) QueueOutcome { // Inspect |flags|, |clock|, and the |partial| sequence to determine an |outcome|. // Keep state mutations *outside* of this if/else block. + // + // Mask off Flag_CONTROL: a control message carries its transaction + // semantics in the low bits (Flag_OUTSIDE_TXN), so it's sequenced and + // committed immediately like any other out-of-transaction message. The + // consumer distinguishes control messages via their retained UUID flags. var outcome QueueOutcome - switch flags { + switch flags &^ Flag_CONTROL { default: w.logError(env, partial, "unexpected UUID flags") fallthrough // Handle as Flag_OUTSIDE_TXN. diff --git a/message/sequencer_test.go b/message/sequencer_test.go index accde9fe..dd7ae03b 100644 --- a/message/sequencer_test.go +++ b/message/sequencer_test.go @@ -438,6 +438,37 @@ func TestSequencerOutsideTxnCases(t *testing.T) { require.Equal(t, pb.Offsets{"test/journal": z2.End}, seq.offsets) } +func TestSequencerControlMessages(t *testing.T) { + var ( + generate = newTestMsgGenerator() + seq = NewSequencer(nil, nil, 0) + A = NewProducerID() + ) + + // A control message carries Flag_OUTSIDE_TXN in its low bits, so it's + // sequenced as an immediately-committed message and dequeues at once. The + // Flag_CONTROL bit is masked off for sequencing rather than tripping the + // unknown-flags path. + var ( + a1 = generate(A, 1, Flag_OUTSIDE_TXN|Flag_CONTROL) + a2 = generate(A, 2, Flag_OUTSIDE_TXN|Flag_CONTROL) + ) + require.Equal(t, []QueueOutcome{QueueOutsideCommit}, queue(seq, a1)) + expectDeque(t, seq, a1) + require.Equal(t, []QueueOutcome{QueueOutsideCommit}, queue(seq, a2)) + expectDeque(t, seq, a2) + require.False(t, seq.HasPending()) + + // The dequeued envelope retains its Flag_CONTROL bit, so a consumer can + // still recognize and drop the control message. + require.NotZero(t, GetFlags(a2.GetUUID())&Flag_CONTROL) + + // A duplicate control message (an already-acked clock) is ignored. + var a1Dup = generate(A, 1, Flag_OUTSIDE_TXN|Flag_CONTROL) + require.Equal(t, []QueueOutcome{QueueOutsideAlreadyAcked}, queue(seq, a1Dup)) + require.Nil(t, seq.emit) +} + func TestSequencerProducerStatesRoundTrip(t *testing.T) { var ( generate = newTestMsgGenerator() diff --git a/message/uuid.go b/message/uuid.go index 55b26fa7..b27820ae 100644 --- a/message/uuid.go +++ b/message/uuid.go @@ -162,6 +162,14 @@ const ( // the progress of the furthest checkpoint ever achieved. Flag_ACK_TXN Flags = 0x2 + // Flag_CONTROL marks the message as an application control message: an + // out-of-band metadata event which carries no content of its own. It + // occupies bit 2 and is orthogonal to the transaction semantics in the + // low two bits: the Sequencer masks Flag_CONTROL off and sequences the + // message by its underlying Flag_OUTSIDE_TXN / Flag_CONTINUE_TXN / + // Flag_ACK_TXN, while consumers inspect the retained bit to recognize it. + Flag_CONTROL Flags = 0x4 + // g1582ns100 is the time interval between 15 Oct 1582 (RFC 4122) // and 1 Jan 1970 (Unix epoch), in units of 100 nanoseconds. g1582ns100 = 122192928000000000 @@ -169,14 +177,19 @@ const ( // String returns a string representation of the Flags value. func (f Flags) String() string { - switch f { + var s string + switch f &^ Flag_CONTROL { case Flag_OUTSIDE_TXN: - return "OUTSIDE_TXN" + s = "OUTSIDE_TXN" case Flag_ACK_TXN: - return "ACK_TXN" + s = "ACK_TXN" case Flag_CONTINUE_TXN: - return "CONTINUE_TXN" + s = "CONTINUE_TXN" default: return fmt.Sprintf("Flags(%x)", uint16(f)) } + if f&Flag_CONTROL != 0 { + s += "|CONTROL" + } + return s } diff --git a/message/uuid_test.go b/message/uuid_test.go index d95882cc..2962ad32 100644 --- a/message/uuid_test.go +++ b/message/uuid_test.go @@ -81,3 +81,16 @@ func TestUUIDBuilding(t *testing.T) { // Original nanos (981273734) rounded to 100ns (981273700) + 2000ns from 2 ticks = 981275700 require.Equal(t, int64(981275700), nsec) } + +func TestFlagsString(t *testing.T) { + require.Equal(t, "OUTSIDE_TXN", Flag_OUTSIDE_TXN.String()) + require.Equal(t, "CONTINUE_TXN", Flag_CONTINUE_TXN.String()) + require.Equal(t, "ACK_TXN", Flag_ACK_TXN.String()) + + // The control bit composes with the underlying transaction flag. + require.Equal(t, "OUTSIDE_TXN|CONTROL", Flag_CONTROL.String()) + require.Equal(t, "CONTINUE_TXN|CONTROL", (Flag_CONTINUE_TXN | Flag_CONTROL).String()) + + // Genuinely-unrecognized flags fall back to hex. + require.Equal(t, "Flags(100)", Flags(0x100).String()) +}