Function TestRequeueOnError

modules/livestore/live_store_test.go:932–970

(t *testing.T)

}

func TestRequeueOnError(t *testing.T) {
	tmpDir := t.TempDir()

	cfg := defaultConfig(t, tmpDir)
	initialBackoff := 100 * time.Millisecond
	cfg.initialBackoff = initialBackoff
	cfg.maxBackoff = 3 * initialBackoff
	cfg.CompleteBlockConcurrency = 1 // to simplify the test
	cfg.holdAllBackgroundProcesses = false
	liveStore, err := liveStoreWithConfig(t, cfg)
	require.NoError(t, err)
	require.NotNil(t, liveStore)

	inst, err := liveStore.getOrCreateInstance(testTenantID)
	require.NoError(t, err)
	enc := erroredEnc{
		VersionedEncoding: inst.completeBlockEncoding,
		mx:                sync.Mutex{},
	}
	enc.SetError(errors.New("forced error"))
	inst.completeBlockEncoding = &enc

	// push data
	expectedID, expectedTrace := pushToLiveStore(t, liveStore)
	requireTraceInLiveStore(t, liveStore, expectedID, expectedTrace)
	requireInstanceState(t, inst, instanceState{liveTraces: 1, walBlocks: 0, completeBlocks: 0})

	// cut to wal and enqueue complete operation
	liveStore.cutAllInstancesToWal()
	requireInstanceState(t, inst, instanceState{liveTraces: 0, walBlocks: 1, completeBlocks: 0})

	// wait for the first backoff that should not be successful
	time.Sleep(initialBackoff * 2)
	requireInstanceState(t, inst, instanceState{liveTraces: 0, walBlocks: 1, completeBlocks: 0})
	// now completeBlockEncoding does not error and block should be flushed successfully
	enc.SetError(nil)
	time.Sleep(initialBackoff * 8)
	requireInstanceState(t, inst, instanceState{liveTraces: 0, walBlocks: 0, completeBlocks: 1})
}

type instanceState struct {
	liveTraces     int