Function TestContinuousChangesBackfill

db/change_cache_test.go:578–676

Test backfill of late arriving sequences to a continuous changes feed

(t *testing.T)

// Test backfill of late arriving sequences to a continuous changes feed
func TestContinuousChangesBackfill(t *testing.T) {

	base.SetUpTestLogging(t, base.LevelInfo, base.KeyCache, base.KeyChanges, base.KeyDCP)

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

	// Create a user with access to channel ABC
	authenticator := db.Authenticator(ctx)
	user, err := authenticator.NewUser("naomi", "letmein", channels.BaseSetOf(t, "ABC", "PBS", "NBC", "CBS"))
	require.NoError(t, 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{"PBS"}, 5)
	WriteDirect(t, collection, []string{"CBS"}, 6)

	db.user, err = authenticator.GetUser("naomi")
	require.NoError(t, err)

	// Start changes feed
	var options ChangesOptions
	options.Since = SequenceID{Seq: 0}
	ctx, changesCtxCancel := context.WithCancel(ctx)
	options.ChangesCtx = ctx
	options.Continuous = true
	options.Wait = true
	defer changesCtxCancel()

	dbCollection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
	feed, err := dbCollection.MultiChangesFeed(ctx, base.SetOf("*"), options)
	assert.True(t, err == nil)

	time.Sleep(50 * time.Millisecond)

	collection = dbCollection.DatabaseCollection
	// Write some more docs
	WriteDirect(t, collection, []string{"CBS"}, 3)
	WriteDirect(t, collection, []string{"PBS"}, 12)
	require.NoError(t, dbCollection.changeCache().waitForSequence(ctx, 12, base.DefaultWaitForSequence))

	// Test multiple backfill in single changes loop iteration
	WriteDirect(t, collection, []string{"ABC", "NBC", "PBS", "CBS"}, 4)
	WriteDirect(t, collection, []string{"ABC", "NBC", "PBS", "CBS"}, 7)
	WriteDirect(t, collection, []string{"ABC", "PBS"}, 8)
	WriteDirect(t, collection, []string{"ABC", "PBS"}, 13)
	require.NoError(t, dbCollection.changeCache().waitForSequence(ctx, 13, base.DefaultWaitForSequence))
	time.Sleep(50 * time.Millisecond)

	// We can't guarantee how compound sequences will be generated in a multi-core test - will
	// depend on timing of arrival in late sequence logs.  e.g. could come through as any one of
	// the following (where all are valid), depending on timing:
	//  ..."4","7","8","8::13"
	//  ..."4", "6::7", "6::8", "6::13"