Function TestLowSequenceHandling

db/change_cache_test.go:679–741

Test low sequence handling of late arriving sequences to a continuous changes feed

(t *testing.T)

// Test low sequence handling of late arriving sequences to a continuous changes feed
func TestLowSequenceHandling(t *testing.T) {
	base.SetUpTestLogging(t, base.LevelDebug, base.KeyCache, base.KeyChanges, base.KeyQuery)

	db, ctx := setupTestDBWithCacheOptions(t, shortWaitCache())
	defer db.Close(ctx)

	// Create a user with access to channel ABC
	authenticator := db.Authenticator(ctx)
	assert.True(t, authenticator != nil, "db.Authenticator(ctx) returned nil")
	user, err := authenticator.NewUser("naomi", "letmein", channels.BaseSetOf(t, "ABC", "PBS", "NBC", "TBS"))
	assert.NoError(t, err, fmt.Sprintf("Error creating new user: %v", err))
	require.NoError(t, authenticator.Save(user))

	collection := GetSingleDatabaseCollection(t, db.DatabaseContext)
	// Simulate seq 3 and 4 being delayed - write 1,2,5,6
	WriteDirect(t, collection, []string{"ABC", "NBC"}, 1)
	WriteDirect(t, collection, []string{"ABC"}, 2)
	WriteDirect(t, collection, []string{"ABC", "PBS"}, 5)
	WriteDirect(t, collection, []string{"ABC", "PBS"}, 6)

	dbCollection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
	require.NoError(t, dbCollection.changeCache().waitForSequence(ctx, 6, base.DefaultWaitForSequence))
	dbCollection.user, err = authenticator.GetUser("naomi")
	require.NoError(t, err)

	// Start changes feed

	var options ChangesOptions
	options.Since = SequenceID{Seq: 0}
	changesCtx, changesCtxCancel := context.WithCancel(base.TestCtx(t))
	options.ChangesCtx = changesCtx
	defer changesCtxCancel()
	options.Continuous = true
	options.Wait = true
	feed, err := dbCollection.MultiChangesFeed(ctx, base.SetOf("*"), options)
	assert.True(t, err == nil)

	changes, err := verifySequencesInFeed(feed, []uint64{1, 2, 5, 6})
	assert.True(t, err == nil)
	require.Len(t, changes, 4)

	collectionID := collection.GetCollectionID()

	require.Equal(t, &ChangeEntry{
		Seq:          SequenceID{Seq: 1, TriggeredBy: 0, LowSeq: 2},
		ID:           "doc-1",
		Changes:      []ChangeByVersionType{{"rev": "1-a"}},
		collectionID: collectionID}, changes[0])

	// Test backfill clear - sequence numbers go back to standard handling
	WriteDirect(t, collection, []string{"ABC", "NBC", "PBS", "TBS"}, 3)
	WriteDirect(t, collection, []string{"ABC", "PBS"}, 4)

	_, err = verifySequencesInFeed(feed, []uint64{3, 4})
	assert.True(t, err == nil)

	WriteDirect(t, collection, []string{"ABC"}, 7)
	WriteDirect(t, collection, []string{"ABC", "NBC"}, 8)