Make flush asynchronous in kv_cached_file.go

[ihtefyw]

KVCachedFile currently hangs when the flushing threshold is reached, waiting for all the values to be written down to disk. This is expensive, especially if used for components active during block processing.
This PR makes the flush asynchronous, unblocking the user immediately after setting an element (conditionally).

The maxPendingFlushes threshold poses a limit to the memory pressure of the asynchronous updates, and adding a new element will wait until the number of elements to flush is lower than the upper bound.
It is set to flushBufferThreshold + 1 in order to allow at least one completely asynchronous flush of the flush buffer

[codecov]

Codecov Report

❌ Patch coverage is 91.66667% with 11 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
go/backend/kv_file/kv_cached_file.go	91.66%	6 Missing and 5 partials ⚠️

Files with missing lines	Coverage Δ
go/backend/kv_file/kv_cached_file.go	88.09% <91.66%> (+2.62%)	⬆️

... and 2 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

This PR updates KVCachedFile to avoid blocking callers when the flush-buffer threshold is reached by moving disk persistence into a background writer goroutine, while keeping Flush, Iterate, and Close as durability barriers.

Changes:

• Introduced an asynchronous flush worker with a bounded pending-queue (maxPendingFlushes) to provide back-pressure.
• Updated read paths (Get/Has) to consult queued (in-flight) buffers so reads can observe recently written values before they reach disk.
• Expanded and adapted tests to validate async error surfacing, FIFO buffer persistence, back-pressure behavior, and race-freedom.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
go/backend/kv_file/kv_cached_file.go	Adds async background flushing with pending-buffer queueing and sticky error handling.
go/backend/kv_file/kv_cached_file_test.go	Updates/extends tests to cover async flush behavior, ordering, back-pressure, and concurrency.

Comments suppressed due to low confidence (4)

go/backend/kv_file/kv_cached_file_test.go:338

• This test blocks the background writer on release. If an assertion fails before close(release), the shutdownWriter() cleanup can hang waiting for the writer goroutine. Add a t.Cleanup that safely closes release if it wasn’t closed yet.

	// Seal a buffer for the background writer and block it inside SetBatch so
	// the buffer stays in flight while we re-Set one of its keys.
	release := make(chan struct{})
	firstWrite := make(chan struct{})
	var once sync.Once

go/backend/kv_file/kv_cached_file_test.go:720

• This test blocks the background writer on release. If an assertion fails before close(release), the shutdownWriter() cleanup can hang waiting for the writer goroutine. Add a t.Cleanup that safely closes release if it wasn’t closed yet.

	// Block the background writer on the first buffer so the queue cannot drain.
	release := make(chan struct{})
	writing := make(chan struct{})
	var once sync.Once
	mock.EXPECT().SetBatch(gomock.Any()).DoAndReturn(func(map[K]V) error {

go/backend/kv_file/kv_cached_file_test.go:968

• This test blocks the background writer on release. If an assertion fails before close(release), the shutdownWriter() cleanup can hang waiting for the writer goroutine. Add a t.Cleanup that safely closes release if it wasn’t closed yet.

	// Block the writer inside SetBatch so a sealed buffer stays in c.pending; the
	// writer keeps the buffer it is writing in the queue until the write returns,
	// so both sealed buffers coexist there while we read the key.
	release := make(chan struct{})
	writing := make(chan struct{})
	var once sync.Once
	mock.EXPECT().SetBatch(gomock.Any()).DoAndReturn(func(map[K]V) error {

go/backend/kv_file/kv_cached_file_test.go:1007

• This test blocks the background writer on release. If an assertion fails before close(release), the shutdownWriter() cleanup can hang waiting for the writer goroutine. Add a t.Cleanup that safely closes release if it wasn’t closed yet.

	c, mock := openTestKVCachedFile(t)

	release := make(chan struct{})
	writing := make(chan struct{})
	var once sync.Once
	var writeOrder []V
	mock.EXPECT().SetBatch(gomock.Any()).DoAndReturn(func(entries map[K]V) error {

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[HerbertJordan]

The implementation looks good, but the tests are to complex for me to think through in reasonable time. As per your suggestion, I rely on you to have worked them out with proper care.