test(s3/lifecycle): pin Schedule edge cases beyond happy path (Phase 15 slice) (#9403)

* 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.

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()
+}