message: add CONTROL flag for out-of-band metadata messages

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.

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()

message/uuid.go

@@ -162,21 +162,34 @@ 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
 )
 
 // 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
 }

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())
+}