From 619cb3982749bdaa82c37b70fd844c6dcb829251 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 9 May 2026 20:35:22 -0700 Subject: [PATCH] test(s3/lifecycle): pin Schedule edge cases beyond happy path (Phase 15 slice) (#9403) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(s3/lifecycle): pin Schedule edge cases beyond happy path Pre-existing schedule_test covered the happy path (ordered Drain, empty schedule, duplicates, boundary-inclusive). Five new tests pin edge cases the dispatcher relies on: - Drain at a time before any DueTime returns nil and leaves the heap intact, so the dispatcher can't accidentally consume future-due matches. - NextDue after partial Drain points to the next earliest, catching a Drain that forgets the heap invariant. - Add after Drain bubbles a fresh earlier DueTime to the front, so late-arriving high-priority matches don't sit behind older ones. - Drain returns Matches in ascending DueTime order regardless of insert order — explicit pinning of the documented contract. - Concurrent Add+Drain across 64 goroutines under -race. * test(s3/lifecycle): actually exercise Drain in AddAfterDrain test Per coderabbit review on #9403: the test name promised "after Drain" but the previous body only Add'd both items without ever calling Drain in between. Insert a real Drain (popping "drain_me") before the second Add, so the heap-invariant-across-Drain-then-Add path is actually pinned. Bumps the after-Drain Match's DueTime out of the way so the Drain in step 3 returns it deterministically. --- .../s3api/s3lifecycle/router/schedule_test.go | 117 ++++++++++++++++++ 1 file changed, 117 insertions(+) diff --git a/weed/s3api/s3lifecycle/router/schedule_test.go b/weed/s3api/s3lifecycle/router/schedule_test.go index f6cd0827ba..5ec4aa12b7 100644 --- a/weed/s3api/s3lifecycle/router/schedule_test.go +++ b/weed/s3api/s3lifecycle/router/schedule_test.go @@ -1,6 +1,7 @@ package router import ( + "sync" "testing" "time" @@ -84,3 +85,119 @@ func TestScheduleAllowsDuplicates(t *testing.T) { t.Fatalf("Drain dup count=%d, want 2", len(got)) } } + +func TestScheduleDrainBeforeAnyDueReturnsNothing(t *testing.T) { + // Drain at a time before the earliest DueTime must return an empty + // slice and leave the heap intact. Otherwise the dispatcher would + // consume future-due matches early. + s := NewSchedule() + t0 := time.Now() + s.Add(mkMatch(t0.Add(5*time.Second), "a")) + s.Add(mkMatch(t0.Add(10*time.Second), "b")) + + got := s.Drain(t0) + if got != nil { + t.Fatalf("Drain before any due should be nil, got %+v", got) + } + if s.Len() != 2 { + t.Fatalf("Len after no-op Drain=%d, want 2", s.Len()) + } +} + +func TestScheduleNextDueAfterPartialDrain(t *testing.T) { + // After draining a prefix, NextDue must point at the earliest + // remaining Match — regression catch for a Drain implementation + // that forgets to maintain the heap invariant. + s := NewSchedule() + t0 := time.Now() + s.Add(mkMatch(t0.Add(1*time.Second), "a")) + s.Add(mkMatch(t0.Add(2*time.Second), "b")) + s.Add(mkMatch(t0.Add(3*time.Second), "c")) + + got := s.Drain(t0.Add(1500 * time.Millisecond)) + if len(got) != 1 || got[0].ObjectKey != "a" { + t.Fatalf("Drain prefix=%+v, want [a]", got) + } + due, ok := s.NextDue() + if !ok || !due.Equal(t0.Add(2*time.Second)) { + t.Fatalf("NextDue after partial Drain=%v ok=%v, want t+2s", due, ok) + } +} + +func TestScheduleAddAfterDrainKeepsOrder(t *testing.T) { + // Adding to a non-empty schedule mid-stream — after a real Drain + // has popped at least one entry, a fresh Match with an earlier + // DueTime than the existing minimum must become the next drainable. + // Pins that heap.Pop + Push together preserve the heap invariant + // across the Drain → Add boundary. + s := NewSchedule() + t0 := time.Now() + s.Add(mkMatch(t0.Add(10*time.Second), "old")) + s.Add(mkMatch(t0.Add(6*time.Second), "drain_me")) + + drained := s.Drain(t0.Add(7 * time.Second)) + if len(drained) != 1 || drained[0].ObjectKey != "drain_me" { + t.Fatalf("pre-drain=%+v, want [drain_me]", drained) + } + + s.Add(mkMatch(t0.Add(5*time.Second), "new")) // added after Drain, earlier than old + + due, ok := s.NextDue() + if !ok || !due.Equal(t0.Add(5*time.Second)) { + t.Fatalf("NextDue=%v ok=%v, want t+5s", due, ok) + } + + got := s.Drain(t0.Add(6 * time.Second)) + if len(got) != 1 || got[0].ObjectKey != "new" { + t.Fatalf("Drain=%+v, want [new]", got) + } +} + +func TestScheduleDrainOrderIsAscendingDueTime(t *testing.T) { + // Drain must return Matches in DueTime order regardless of insert + // order — explicit pinning of the contract documented on Drain. + s := NewSchedule() + t0 := time.Now() + for _, off := range []time.Duration{ + 5 * time.Second, + 1 * time.Second, + 3 * time.Second, + 2 * time.Second, + 4 * time.Second, + } { + s.Add(mkMatch(t0.Add(off), "k")) + } + + got := s.Drain(t0.Add(10 * time.Second)) + if len(got) != 5 { + t.Fatalf("Drain count=%d, want 5", len(got)) + } + for i := 1; i < len(got); i++ { + if got[i].DueTime.Before(got[i-1].DueTime) { + t.Fatalf("Drain[%d]=%v is before Drain[%d]=%v", i, got[i].DueTime, i-1, got[i-1].DueTime) + } + } +} + +func TestScheduleConcurrentAddDrainNoRace(t *testing.T) { + // The dispatcher's Add and Drain run on separate goroutines; the + // schedule's mutex must serialize them without deadlock. -race + // catches a regression that drops the lock on either path. + s := NewSchedule() + t0 := time.Now() + const N = 64 + var wg sync.WaitGroup + wg.Add(N * 2) + for i := 0; i < N; i++ { + i := i + go func() { + defer wg.Done() + s.Add(mkMatch(t0.Add(time.Duration(i)*time.Millisecond), "k")) + }() + go func() { + defer wg.Done() + _ = s.Drain(t0.Add(10 * time.Second)) + }() + } + wg.Wait() +}