Function TestChannelRace

db/change_cache_test.go:1025–1121

Test race condition causing skipped sequences in changes feed. Channel feeds are processed sequentially in the main changes.go iteration loop, without a lock on the underlying channel caches. The following sequence is possible while running a changes feed for channels "A", "B": 1. Sequence 100, Ch

(t *testing.T)

//	base.Infof(base.KeyChanges, "Simulate slow processing time for channel %s - sleeping for 100 ms", channel)
//	time.Sleep(100 * time.Millisecond)
func TestChannelRace(t *testing.T) {
	base.SetUpTestLogging(t, base.LevelInfo, base.KeyChanges)

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

	// Create a user with access to channels "Odd", "Even"
	authenticator := db.Authenticator(ctx)
	user, err := authenticator.NewUser("naomi", "letmein", channels.BaseSetOf(t, "Even", "Odd"))
	require.NoError(t, err)
	require.NoError(t, authenticator.Save(user))

	collection := GetSingleDatabaseCollection(t, db.DatabaseContext)

	// Write initial sequences
	WriteDirect(t, collection, []string{"Odd"}, 1)
	WriteDirect(t, collection, []string{"Even"}, 2)
	WriteDirect(t, collection, []string{"Odd"}, 3)

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

	// Start changes feed
	dbCollection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)

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

	// Go-routine to work the feed channel and write to an array for use by assertions
	var changes struct {
		lock    sync.RWMutex
		entries []*ChangeEntry
	}
	changes.entries = make([]*ChangeEntry, 0, 50)
	go func() {
		for feedClosed == false {
			select {
			case entry, ok := <-feed:
				if ok {
					// feed sends nil after each continuous iteration
					if entry != nil {
						log.Println("Changes entry:", entry.Seq)
						changes.lock.Lock()
						changes.entries = append(changes.entries, entry)
						changes.lock.Unlock()
					}
				} else {
					log.Println("Closing feed")
					feedClosed = true
				}