Function TestVariableRateAllocators

db/sequence_allocator_test.go:767–848

TestVariableRateAllocators simulates the following scenario: - import nodes have high sequence allocation rate - client-facing nodes have low sequence allocation rate - documents are imported, then the same documents are immediately updated by clients (including sequence validation triggering nextSe

(t *testing.T)

//
// Ensures we don't release more sequences than would be expected based on allocator batch size
func TestVariableRateAllocators(t *testing.T) {
	ctx := base.TestCtx(t)
	bucket := base.GetTestBucket(t)
	defer bucket.Close(ctx)
	var expectedAllocations uint64

	dataStore := bucket.GetSingleDataStore()
	stats, err := base.NewSyncGatewayStats()
	require.NoError(t, err)

	importStats, err := stats.NewDBStats("import", false, false, false, nil, nil)
	require.NoError(t, err)

	importFeedAllocator, err := newSequenceAllocator(ctx, dataStore, importStats.DatabaseStats, base.DefaultMetadataKeys)
	require.NoError(t, err)

	// All test allocators are stopped when allocatorCtx is closed
	allocatorCtx, cancelFunc := context.WithCancel(ctx)

	// Start import node allocator, performing 10000 allocations/second.
	var allocatorWg sync.WaitGroup
	allocatorWg.Add(1)
	go func() {
		count := runAllocator(allocatorCtx, importFeedAllocator, 100*time.Microsecond) // 10000 writes/second
		atomic.AddUint64(&expectedAllocations, count)
		allocatorWg.Done()
	}()

	// Start multiple client node allocators, performing 100 allocations/second
	clientAllocators := make([]*sequenceAllocator, 0)
	clientAllocatorCount := 10
	for i := 0; i <= clientAllocatorCount; i++ {
		clientStats, err := stats.NewDBStats(fmt.Sprintf("client%d", i), false, false, false, nil, nil)
		require.NoError(t, err)
		clientAllocator, err := newSequenceAllocator(ctx, dataStore, clientStats.DatabaseStats, base.DefaultMetadataKeys)
		require.NoError(t, err)
		clientAllocators = append(clientAllocators, clientAllocator)
		allocatorWg.Add(1)
		go func() {
			count := runAllocator(allocatorCtx, clientAllocator, 10*time.Millisecond) // 100 writes/second
			atomic.AddUint64(&expectedAllocations, count)
			allocatorWg.Done()
		}()
	}

	// Wait for allocators to get up to maximum batch size
	time.Sleep(500 * time.Millisecond)
	documentCount := 10
	var updateWg sync.WaitGroup
	updateWg.Add(documentCount)
	for i := 0; i < documentCount; i++ {
		go func() {
			_ = multiNodeUpdate(t, ctx, importFeedAllocator, clientAllocators, 5, 10*time.Millisecond)
			updateWg.Done()
			atomic.AddUint64(&expectedAllocations, 6)
		}()
	}