mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-19 05:36:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: wait for every queued flush before deleting the log files (#10546)

Browse Source 
TestSubscribeLoop_FlushProvenGapSkipsToRetained deleted the log files
once the eviction watermark's own flush had landed, while the windows
sealed after it were still queued. Those flushes then wrote their files
back, and the subscriber served them from disk instead of taking the
gap-skip path, so the windows whose files really were gone came out
missing. Wait through the last sealed window instead.
This commit is contained in:
Chris Lu
2026-08-03 09:26:42 -07:00
committed by GitHub
parent 4f692bf9c3
commit a9de90ae29
1 changed files with 14 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/server/filer_subscribe_loop_test.go
+14 -5
View File
@@ -507,10 +507,16 @@ func TestSubscribeLoop_BoundedSubscriptionTerminates(t *testing.T) {
func TestSubscribeLoop_FlushProvenGapSkipsToRetained(t *testing.T) {
h := newSubscribeHarness(t)
for w := 0; w < log_buffer.PreviousBufferCount+2; w++ {
windows := log_buffer.PreviousBufferCount + 2
for w := 0; w < windows; w++ {
h.append(h.tsAt(w, 0))
}
waitForFlushedFiles(t, h)
// The last append seals the window before it, so every window but the
// current one is queued for flush. Wait for all of them, not just the one
// the eviction watermark names: a flush still in flight writes its file
// back after the delete below, and the subscriber then serves that window
// from disk instead of taking the gap path this test is about.
waitForFlushedFiles(t, h, h.tsAt(windows-2, 0))
if h.f.LocalMetaLogBuffer.GetLastEvictedTsNs() == 0 {
t.Fatal("precondition: nothing evicted")
}
@@ -518,7 +524,7 @@ func TestSubscribeLoop_FlushProvenGapSkipsToRetained(t *testing.T) {
earliest := h.f.LocalMetaLogBuffer.GetEarliestTime().UnixNano()
var retained []int64
for w := 0; w < log_buffer.PreviousBufferCount+2; w++ {
for w := 0; w < windows; w++ {
if ts := h.tsAt(w, 0); ts >= earliest {
retained = append(retained, ts)
}
@@ -528,11 +534,14 @@ func TestSubscribeLoop_FlushProvenGapSkipsToRetained(t *testing.T) {
waitForEvents(t, r, retained, 5*time.Second)
}
func waitForFlushedFiles(t *testing.T, h *subscribeHarness) {
// waitForFlushedFiles waits until the flush of the window ending at
// throughTsNs has landed. Flushes run in queue order on one goroutine, so that
// also proves every window sealed before it is on disk and none is in flight.
func waitForFlushedFiles(t *testing.T, h *subscribeHarness, throughTsNs int64) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if h.f.LocalMetaLogBuffer.GetLastFlushTsNs() >= h.f.LocalMetaLogBuffer.GetLastEvictedTsNs() {
if h.f.LocalMetaLogBuffer.GetLastFlushTsNs() >= throughTsNs {
return
}
time.Sleep(10 * time.Millisecond)