Function TestWatchRemoteKVStore

pkg/allocator/allocator_test.go:528–661

(t *testing.T)

}

func TestWatchRemoteKVStore(t *testing.T) {
	var wg sync.WaitGroup
	var synced atomic.Bool

	run := func(ctx context.Context, rc RemoteIDCache) context.CancelFunc {
		ctx, cancel := context.WithCancel(ctx)
		wg.Go(func() {
			rc.Watch(ctx, func(context.Context) { synced.Store(true) })
		})
		return cancel
	}

	stop := func(cancel context.CancelFunc) {
		cancel()
		wg.Wait()
		synced.Store(false)
	}

	global := Allocator{logger: hivetest.Logger(t), remoteCaches: make(map[string]*remoteCache)}
	events := make(AllocatorEventChan, 10)

	ctx, cancel := context.WithCancel(context.Background())

	// Ensure that the goroutines are properly collected also in case the test fails.
	defer stop(cancel)

	newRemoteAllocator := func(backend Backend) *Allocator {
		remote, err := NewAllocator(hivetest.Logger(t), TestAllocatorKey(""), backend, WithEvents(events), WithoutAutostart(), WithoutGC())
		require.NoError(t, err)

		return remote
	}

	// Add a new remote cache, and assert that it is registered correctly
	// and the proper events are emitted
	backend := newDummyBackend()
	remote := newRemoteAllocator(backend)

	backend.AllocateID(ctx, idpool.ID(1), TestAllocatorKey("foo"))
	backend.AllocateID(ctx, idpool.ID(2), TestAllocatorKey("baz"))

	rc := global.NewRemoteCache("remote", remote)
	require.False(t, rc.Synced(), "The cache should not be synchronized")
	cancel = run(ctx, rc)

	require.Equal(t, AllocatorEvent{ID: idpool.ID(1), Key: TestAllocatorKey("foo"), Typ: AllocatorChangeUpsert}, <-events)
	require.Equal(t, AllocatorEvent{ID: idpool.ID(2), Key: TestAllocatorKey("baz"), Typ: AllocatorChangeUpsert}, <-events)

	ev := <-events
	require.Equal(t, AllocatorChangeSync, ev.Typ)
	require.False(t, rc.Synced(), "The cache should not be synchronized")
	close(ev.Done)

	require.Eventually(t, func() bool {
		global.remoteCachesMutex.RLock()
		defer global.remoteCachesMutex.RUnlock()
		return global.remoteCaches["remote"] == rc
	}, 1*time.Second, 10*time.Millisecond)