fix: drain canceled storage commits before releasing object locks

Wait for admitted data and metadata mutations, preserve overwrite backups during rollback, and reject rollback of a different generation. Report conflicting or unreadable pool metadata as read-quorum failures instead of exposing stale copies. Add barrier-driven local/RPC and multi-pool regressions for cancellation, self-copy, rollback and old-value reads.

Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-10-25 00:19:42 +08:00
parent 4f17a23e33
commit cd1f6cec3c
9 changed files with 730 additions and 25 deletions
+15
View File
@@ -127,6 +127,17 @@ var deadlineWorkPanics atomic.Int64
// If the deadline/context passes before the function finishes executing,
// the zero value and the context error is returned.
func WithDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) {
return withDeadline(ctx, timeout, work, false)
}
// WithDeadlineAndWait cancels work at the deadline, but waits for it to finish
// and returns its actual result. Use it for mutations whose caller must not
// release a lock or start cleanup while the work can still change storage.
func WithDeadlineAndWait[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) {
return withDeadline(ctx, timeout, work, true)
}
func withDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error), wait bool) (result V, err error) {
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
@@ -163,6 +174,10 @@ func WithDeadline[V any](ctx context.Context, timeout time.Duration, work func(c
case v := <-c:
return v.val, v.err
case <-ctx.Done():
if wait {
v := <-c
return v.val, v.err
}
var zero V
return zero, ctx.Err()
}
+53
View File
@@ -25,9 +25,62 @@ import (
"os"
"strings"
"testing"
"testing/synctest"
"time"
)
func TestWithDeadlineAndWait(t *testing.T) {
for _, deadline := range []bool{false, true} {
name := "cancel"
if deadline {
name = "deadline"
}
t.Run(name, func(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
entered, release := make(chan struct{}), make(chan struct{})
done := make(chan ioret[int], 1)
go func() {
v, err := WithDeadlineAndWait(ctx, time.Second, func(context.Context) (int, error) {
close(entered)
<-release
return 42, nil
})
done <- ioret[int]{v, err}
}()
<-entered
if deadline {
time.Sleep(2 * time.Second)
} else {
cancel()
}
synctest.Wait()
select {
case r := <-done:
t.Fatalf("returned before mutation completed: %+v", r)
default:
}
close(release)
synctest.Wait()
r := <-done
if r.err != nil || r.val != 42 {
t.Fatalf("lost actual commit result: %+v", r)
}
})
})
}
}
func TestWithDeadlineAndWaitPanic(t *testing.T) {
_, err := WithDeadlineAndWait(t.Context(), time.Second, func(context.Context) (int, error) {
panic("mutation failure")
})
if err == nil || !strings.Contains(err.Error(), "mutation failure") {
t.Fatalf("worker panic was not returned: %v", err)
}
}
type sleepWriter struct {
timeout time.Duration
}