add test to ensure no negative interaction between changes feed and compound sequences.

Author: RIT3shSapata

rest/changestest/changes_api_test.go

@@ -3761,6 +3761,145 @@ func TestOneShotGrantRequestPlusDbConfig(t *testing.T) {
 	oneShotComplete.Wait()
 }
 
+// TestMultiChannelChangesWithTriggeredSequence tests that the changes feed correctly handles
+// a compound since value containing both LowSeq and TriggeredBy (e.g. "6:10:3"), which is
+// produced when a channel access grant (backfill, setting TriggeredBy) occurs while LowSeq
+// mode is active (due to skipped sequences).
+//
+// This exercises the corner case fixed by CBG-5429: before the fix, SequenceID.String() would
+// silently drop the LowSeq field when LowSeq > Seq, causing the client to receive an incorrect
+// last_seq. On the next request the changes feed would use the wrong since value, potentially
+// skipping backfill documents.
+//
+// Sequence layout written by WriteDirect (seq 1 = user creation via CreateUser):
+//
+//	Seq  Channel  Notes
+//	1    (user)   CreateUser allocates this via the sequence allocator
+//	2    ABC
+//	3    DEF      not visible until DEF access is granted
+//	4    DEF      not visible until DEF access is granted
+//	5    ABC
+//	6    ABC
+//	7    DEF      written late (after 9) to create the skipped-sequence / LowSeq condition
+//	8    (gap)    never written; skipped sequence
+//	9    ABC
+//	10+  (user update granting DEF access, after allocator is advanced)
+func TestMultiChannelChangesWithTriggeredSequence(t *testing.T) {
+	// base.SetUpTestLogging(t, base.LevelDebug, base.KeyChanges, base.KeyCache, base.KeyHTTP)
+
+	// MaxWaitPending: reduce from the 5 s default to 5 ms so the cache promotes pending
+	// sequences to the skipped queue almost immediately, keeping test latency low.
+	pendingMaxWait := uint32(5)
+
+	rt := rest.NewRestTester(t, &rest.RestTesterConfig{
+		SyncFn: `function(doc, oldDoc) {channel(doc.channels);}`,
+		DatabaseConfig: &rest.DatabaseConfig{
+			DbConfig: rest.DbConfig{
+				CacheConfig: &rest.CacheConfig{
+					ChannelCacheConfig: &rest.ChannelCacheConfig{
+						MaxWaitPending: &pendingMaxWait,
+					},
+				},
+			},
+		},
+	})
+	defer rt.Close()
+	testDb := rt.GetDatabase()
+	ctx := rt.Context()
+
+	// ChannelQueryLimit forces the channel cache to paginate when fetching changes, exercising
+	// the interaction between the per-channel query limit and the request-level limit parameter.
+	testDb.Options.CacheOptions.ChannelQueryLimit = 5
+
+	collection, _ := rt.GetSingleTestDatabaseCollection()
+
+	// CreateUser allocates seq 1 from the sequence allocator (_sync:seq becomes 1).
+	// The user starts with access to ABC only; DEF will be granted later.
+	rt.CreateUser("sg-user", []string{"ABC"})
+
+	// WriteDirect writes documents at explicit SG sequence numbers, bypassing the sequence
+	// allocator entirely (_sync:seq is not updated). This lets us craft a specific sequence
+	// layout including a deliberate gap to trigger LowSeq mode.
+	//
+	// Seq 9 is written before seq 7, creating a gap at seqs 7 and 8. With MaxWaitPending=5ms
+	// the cache quickly moves seqs 7 and 8 into the skipped-sequence queue and promotes seq 9,
+	// setting LowSeq = 6 (last contiguous sequence before the gap).
+	db.WriteDirect(t, collection, []string{"ABC"}, 2)
+	db.WriteDirect(t, collection, []string{"DEF"}, 3)
+	db.WriteDirect(t, collection, []string{"DEF"}, 4)
+	db.WriteDirect(t, collection, []string{"ABC"}, 5)
+	db.WriteDirect(t, collection, []string{"ABC"}, 6)
+	db.WriteDirect(t, collection, []string{"ABC"}, 9) // gap: seqs 7 and 8 are missing
+	// Wait until seq 9 is in the cache and no longer in the skipped queue.
+	rt.WaitForSequenceNotSkipped(9)
+
+	// Verify initial changes feed state: user sees 5 entries (seq 1 user doc, seqs 2/5/6/9 ABC).
+	// DEF docs at seqs 3 and 4 are not visible since user has no DEF access yet.
+	// last_seq = "6::9": LowSeq=6 (seqs 7,8 are in the skipped queue), HighSeq=9.
+	rt.WaitForPendingChanges()
+	changes := rt.GetChanges("/{{.keyspace}}/_changes", "sg-user")
+	require.Len(t, changes.Results, 5)
+	since := changes.Results[0].Seq
+	assert.Equal(t, uint64(1), since.Seq)
+	assert.Equal(t, "6::9", changes.Last_Seq.String())
+
+	// Confirm no spurious results when re-requesting with the compound since value.
+	changesJSON := fmt.Sprintf(`{"since":"%s"}`, changes.Last_Seq.String())
+	changes = rt.PostChanges("/{{.keyspace}}/_changes", changesJSON, "sg-user")
+	require.Len(t, changes.Results, 0)
+
+	// Fill in seq 7 (DEF channel) — one of the two skipped sequences arrives late.
+	// This removes seq 7 from the skipped queue; only seq 8 remains skipped.
+	// LowSeq advances to 7. The late-sequence feeds for this changes iteration will
+	// deliver seq 7 even though it arrived after the initial feed was established.
+	db.WriteDirect(t, collection, []string{"DEF"}, 7)
+	// WaitForSequence(7) confirms the cache has processed past seq 7. Because seq 9 was
+	// already processed earlier (nextSequence=10), this returns immediately.
+	rt.WaitForSequence(7)
+
+	// With seq 7 now in the cache, the changes feed should deliver it via the late-sequence
+	// path. The user still has no DEF access, but the LowSeq advancement itself produces a
+	// result (last_seq moves forward). The last_seq from this response is used as the since
+	// value for the final request below.
+	changes = rt.PostChanges("/{{.keyspace}}/_changes", changesJSON, "sg-user")
+	require.Len(t, changes.Results, 1)
+
+	// WriteDirect bypasses the sequence allocator, so _sync:seq is still at 1 after all the
+	// WriteDirect calls above. If we now do a normal SG write (the user update below) without
+	// advancing the allocator, the write would be assigned seq 2 — colliding with the
+	// WriteDirect doc already at seq 2, and making TriggeredBy=2 for the DEF grant. With
+	// TriggeredBy=2 the backfill would look for DEF docs at seq ≤ 2, missing seqs 3, 4, 7.
+	//
+	// AllocateTestSequence increments _sync:seq directly (without updating the allocator's
+	// local last/max state). After 8 calls _sync:seq = 9. The next normal SG write then
+	// triggers a new batch reservation, landing at seq 10.
+	for range 8 {
+		_, err := db.AllocateTestSequence(ctx, testDb)
+		require.NoError(t, err)
+	}
+
+	// Grant DEF access to the user. This is a normal SG write and allocates seq 10 from the
+	// sequence allocator. Seq 10 becomes the seqAddedAt for the DEF channel grant, and will
+	// appear as TriggeredBy in subsequent changes responses during the backfill.
+	//
+	// Combined with the LowSeq still active (seq 8 remains skipped), the next changes
+	// response will contain a since value of the form {LowSeq: L, TriggeredBy: 10, Seq: S}
+	// — the exact compound format that CBG-5429 fixes.
+	resp := rt.SendAdminRequest(http.MethodPut, "/{{.db}}/_user/sg-user", rest.GetUserPayload(t, "", rest.RestTesterDefaultUserPassword, "", rt.GetSingleDataStore(), []string{"ABC", "DEF"}, nil))
+	rest.RequireStatus(t, resp, http.StatusOK)
+
+	// Use the last_seq from the post-seq7 response as the since value. This since has LowSeq
+	// set, and after the DEF grant the changes feed will add TriggeredBy, producing a compound
+	// since value {LowSeq: L, TriggeredBy: 10, Seq: S} — the CBG-5429 corner case.
+	changesJSON = fmt.Sprintf(`{"since":"%s"}`, changes.Last_Seq.String())
+
+	// Expect 4 results: the user update (seq 10) plus the 3 DEF backfill docs (seqs 3, 4, 7).
+	// If Before() or String() are broken for the compound since value, backfill docs will be
+	// missed and this assertion will fail.
+	changes = rt.PostChanges("/{{.keyspace}}/_changes", changesJSON, "sg-user")
+	require.Len(t, changes.Results, 4)
+}
+
 func waitForCompactStopped(dbc *db.DatabaseContext) error {
 	for range 100 {
 		compactRunning := dbc.CacheCompactActive()