diff --git a/adapter/admin_backup.go b/adapter/admin_backup.go index cf22175ce..6582b9b07 100644 --- a/adapter/admin_backup.go +++ b/adapter/admin_backup.go @@ -196,6 +196,11 @@ type backupSession struct { // deadline, still blocking compaction and capacity for a backup that has // already ended, so renewals are refused from that point on. closing bool + // generation advances on every accepted renewal. A renewal that fails part + // way through its fan-out tears the session down, and that must not undo a + // concurrent renewal that already succeeded: the failing attempt only + // cleans up while the generation it started from is still current. + generation uint64 } type preparedBackup struct { @@ -338,7 +343,17 @@ func (s *AdminServer) pinBackupGroups( ) (map[uint64]uint64, error) { reserveEntry := kv.EncodeBackupReserveEntry(kv.BackupReserveEntry{PinID: pinID, ReadTS: readTS, Deadline: deadline}) if _, _, err := proposeBackupAll(ctx, []backupGroup{controlGroup}, reserveEntry); err != nil { - if backupCapacityReservationFull(err) { + // Only a definitive capacity rejection proves nothing was reserved. + // Every other error is ambiguous -- the entry may well have committed + // with only the response lost, or the context expired after the + // proposal -- and an unacknowledged reservation holds one of the few + // global active-backup slots until its TTL for a backup no caller ever + // received. Unreserve is idempotent and keyed by this pin, so + // compensating costs nothing when the reservation never landed. + if !backupCapacityRejectionDefinitive(err) { + s.compensateBackupRelease(controlGroup, nil, pinID) + } + if backupCapacityRejectionReported(err) { return nil, status.Errorf(codes.ResourceExhausted, "%s", kv.ErrTooManyActiveBackups) } return nil, status.Errorf(codes.Unavailable, "reserve backup capacity: %v", err) @@ -434,20 +449,45 @@ func (s *AdminServer) RenewBackup(ctx context.Context, req *pb.RenewBackupReques if err != nil { return nil, err } + generation, live := s.backupSessionGeneration(tok) deadline, err := s.renewBackupGroups(ctx, groups, tok.pinID, tok.readTS, ttl) if err != nil { - s.compensateBackupRelease(groups[0], groups, tok.pinID) - s.forgetBackupSession(tok.pinID) + s.abandonFailedRenewal(groups, tok, generation, live) return nil, status.Errorf(codes.Unavailable, "renew backup pin: %v", err) } if err := s.requireRenewableBackupToken(tok); err != nil { - s.compensateBackupRelease(groups[0], groups, tok.pinID) - s.forgetBackupSession(tok.pinID) + s.abandonFailedRenewal(groups, tok, generation, live) return nil, err } return s.finishRenewBackup(groups, tok, ttl, deadline) } +// abandonFailedRenewal releases what this attempt may have half-renewed, but +// only while it still owns the session it started from. A renewal that +// overlapped a successful one no longer does, and must leave that caller's +// pins alone. +// +// A session that has *disappeared* is not the same case. EndBackup can remove +// it while this renewal is in flight, and the reserve or partial pin fan-out +// can then commit behind that release and stay active until the new TTL, +// holding one of the few global backup slots and blocking compaction. Only a +// still-live session at a different generation proves another caller owns the +// pins, so that is the one outcome that skips compensation. +func (s *AdminServer) abandonFailedRenewal( + groups []backupGroup, + tok backupToken, + generation uint64, + live bool, +) { + if live && s.forgetBackupSessionAtGeneration(tok.pinID, generation) == backupSessionOwnershipTaken { + return + } + if !live { + s.forgetBackupSession(tok.pinID) + } + s.compensateBackupRelease(groups[0], groups, tok.pinID) +} + func (s *AdminServer) finishRenewBackup( groups []backupGroup, tok backupToken, @@ -932,7 +972,23 @@ func backupProposalGroupError(groupID uint64, err error) error { return errors.Wrapf(err, "raft group %d", groupID) } -func backupCapacityReservationFull(err error) bool { +// backupCapacityRejectionDefinitive reports whether the reserve failure proves +// nothing was reserved, which is the only case where skipping the compensating +// unreserve is safe. Only a local apply response carries +// kv.ErrTooManyActiveBackups as a Go error. backupProposalGroupError re-stamps +// *any* ResourceExhausted status -- including one produced by a transport, +// proxy, or server-side quota layer -- so a bare code is ambiguous and must be +// compensated. +func backupCapacityRejectionDefinitive(err error) bool { + return errors.Is(err, kv.ErrTooManyActiveBackups) +} + +// backupCapacityRejectionReported reports whether the caller should see +// ResourceExhausted rather than Unavailable. A forwarded capacity rejection +// loses its Go error across the RPC boundary but keeps its status code, so the +// code still drives the client-facing answer even though it no longer proves +// enough to skip compensation. +func backupCapacityRejectionReported(err error) bool { return errors.Is(err, kv.ErrTooManyActiveBackups) || status.Code(err) == codes.ResourceExhausted } @@ -1174,10 +1230,11 @@ func (s *AdminServer) extendBackupSession(tok backupToken) bool { if !ok || session.closing || session.readTS != tok.readTS { return false } + session.generation++ if tok.deadline.After(session.deadline) { session.deadline = tok.deadline - s.backupSessions[tok.pinID] = session } + s.backupSessions[tok.pinID] = session return true } @@ -1223,6 +1280,57 @@ func (s *AdminServer) requireRenewableBackupToken(tok backupToken) error { return nil } +// backupSessionGeneration reports the generation a renewal is starting from. +// ok is false when there is no live session for the token, in which case there +// is nothing for a failed renewal to tear down. +func (s *AdminServer) backupSessionGeneration(tok backupToken) (uint64, bool) { + s.backupStateMu.Lock() + defer s.backupStateMu.Unlock() + session, ok := s.backupSessions[tok.pinID] + if !ok || session.readTS != tok.readTS { + return 0, false + } + return session.generation, true +} + +// backupSessionOwnership distinguishes the two reasons a generation-guarded +// drop can decline. Only backupSessionOwnershipTaken proves someone else owns +// the pins; backupSessionOwnershipAbsent means EndBackup already removed the +// session, and the caller must still compensate for whatever it half-renewed. +type backupSessionOwnership int + +const ( + // backupSessionOwnershipAbsent: no session at this pin id any more. + backupSessionOwnershipAbsent backupSessionOwnership = iota + // backupSessionOwnershipTaken: a newer generation owns the session. + backupSessionOwnershipTaken + // backupSessionOwnershipDropped: the caller's own session was removed. + backupSessionOwnershipDropped +) + +// forgetBackupSessionAtGeneration drops the session only while it is still the +// one the caller started from. A renewal that failed part way through its +// fan-out must release the pins it may have half-renewed, but if another +// renewal has already succeeded in the meantime it owns the session now -- +// tearing it down here would leave that caller holding a token whose pins are +// gone, with retention free to compact the data underneath it. +func (s *AdminServer) forgetBackupSessionAtGeneration( + pinID kv.BackupPinID, + generation uint64, +) backupSessionOwnership { + s.backupStateMu.Lock() + defer s.backupStateMu.Unlock() + session, ok := s.backupSessions[pinID] + if !ok { + return backupSessionOwnershipAbsent + } + if session.generation != generation { + return backupSessionOwnershipTaken + } + delete(s.backupSessions, pinID) + return backupSessionOwnershipDropped +} + func (s *AdminServer) forgetBackupSession(pinID kv.BackupPinID) { s.backupStateMu.Lock() delete(s.backupSessions, pinID) diff --git a/adapter/admin_backup_renew_race_test.go b/adapter/admin_backup_renew_race_test.go new file mode 100644 index 000000000..0815ab6b1 --- /dev/null +++ b/adapter/admin_backup_renew_race_test.go @@ -0,0 +1,218 @@ +package adapter + +import ( + "context" + stderrors "errors" + "testing" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/bootjp/elastickv/kv" + pb "github.com/bootjp/elastickv/proto" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// A renewal that fails part way through its fan-out releases every group pin +// and forgets the session. If another renewal has already succeeded in the +// meantime, that teardown leaves its caller holding a token whose pins are +// gone, with retention free to compact the versions underneath it -- the +// backup is silently invalid while the caller believes it was renewed. +// +// The concurrent success is injected from inside the failing renewal's own +// fan-out, after it captured the session generation: onPropose fires on the +// reserve that precedes the failing pin, and calls the real +// extendBackupSession, which is what a successful renewal ends with. +func TestRenewBackupKeepsAConcurrentlyRenewedSession(t *testing.T) { + t.Parallel() + + group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000} + proposer := newBackupTestProposer() + srv := newBackupControlTestServer( + t, + &backupTestStore{}, + map[uint64]*backupTestGroup{1: group}, + map[uint64]*backupTestProposer{1: proposer}, + nil, + ) + begin, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{}) + require.NoError(t, err) + tok, err := srv.decodeBackupToken(begin.GetPinToken()) + require.NoError(t, err) + + proposer.mu.Lock() + proposer.failures[backupSubtypePin] = 8 + proposer.transportError[backupSubtypePin] = stderrors.New("leader unavailable") + proposer.onPropose = func(subtype byte, _ uint64) { + if subtype != backupSubtypeReserve { + return + } + // Exactly once, and only after RenewBackup has read the generation it + // will compare against. + proposer.onPropose = nil + require.True(t, srv.extendBackupSession(tok)) + } + proposer.mu.Unlock() + + _, err = srv.RenewBackup(context.Background(), &pb.RenewBackupRequest{PinToken: begin.GetPinToken()}) + require.Equal(t, codes.Unavailable, status.Code(err)) + + // The session the concurrent renewal owns must survive, and no release or + // unreserve may have been proposed on its behalf. + _, err = srv.backupRouteSnapshotForToken(tok) + require.NoError(t, err, "the concurrently renewed session must still be live") + require.NoError(t, srv.requireLiveBackupSession(tok)) + require.NotContains(t, proposer.subtypes(), backupSubtypeRelease) + require.NotContains(t, proposer.subtypes(), backupSubtypeUnreserve) +} + +// With no concurrent renewal the failing attempt still owns the session, so it +// must tear the pin down exactly as before. +func TestRenewBackupStillReleasesWhenItOwnsTheSession(t *testing.T) { + t.Parallel() + + group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000} + proposer := newBackupTestProposer() + srv := newBackupControlTestServer( + t, + &backupTestStore{}, + map[uint64]*backupTestGroup{1: group}, + map[uint64]*backupTestProposer{1: proposer}, + nil, + ) + begin, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{}) + require.NoError(t, err) + tok, err := srv.decodeBackupToken(begin.GetPinToken()) + require.NoError(t, err) + + proposer.mu.Lock() + proposer.failures[backupSubtypePin] = 8 + proposer.transportError[backupSubtypePin] = stderrors.New("leader unavailable") + proposer.mu.Unlock() + + _, err = srv.RenewBackup(context.Background(), &pb.RenewBackupRequest{PinToken: begin.GetPinToken()}) + require.Equal(t, codes.Unavailable, status.Code(err)) + require.Contains(t, proposer.subtypes(), backupSubtypeRelease) + require.Contains(t, proposer.subtypes(), backupSubtypeUnreserve) + _, err = srv.backupRouteSnapshotForToken(tok) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) +} + +// A reservation whose proposal fails for any reason other than a capacity +// rejection is ambiguous: it may have committed with only the response lost. +// Leaving it in place holds one of the few global active-backup slots until +// its TTL for a backup no caller ever received. +func TestBeginBackupUnreservesAmbiguousReservationFailures(t *testing.T) { + t.Parallel() + + group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000} + proposer := newBackupTestProposer() + proposer.failures[backupSubtypeReserve] = 8 + proposer.transportError[backupSubtypeReserve] = stderrors.New("leader unavailable") + srv := newBackupControlTestServer( + t, + &backupTestStore{}, + map[uint64]*backupTestGroup{1: group}, + map[uint64]*backupTestProposer{1: proposer}, + nil, + ) + + _, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{}) + require.Equal(t, codes.Unavailable, status.Code(err)) + require.Contains(t, proposer.subtypes(), backupSubtypeUnreserve, + "an ambiguous reservation must be compensated") +} + +// A session that has *disappeared* is a different case from one a newer +// generation owns. EndBackup can remove it while this renewal is in flight, +// and a reserve or partial pin fan-out that commits behind that release stays +// active until the new TTL -- holding one of the few global backup slots and +// blocking compaction for a backup that has already ended. Only a still-live +// session at another generation proves someone else owns the pins. +func TestRenewBackupCompensatesWhenTheSessionDisappeared(t *testing.T) { + t.Parallel() + + group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000} + proposer := newBackupTestProposer() + srv := newBackupControlTestServer( + t, + &backupTestStore{}, + map[uint64]*backupTestGroup{1: group}, + map[uint64]*backupTestProposer{1: proposer}, + nil, + ) + begin, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{}) + require.NoError(t, err) + tok, err := srv.decodeBackupToken(begin.GetPinToken()) + require.NoError(t, err) + + proposer.mu.Lock() + proposer.failures[backupSubtypePin] = 8 + proposer.transportError[backupSubtypePin] = stderrors.New("leader unavailable") + proposer.onPropose = func(subtype byte, _ uint64) { + if subtype != backupSubtypeReserve { + return + } + // Exactly once, and only after RenewBackup captured the generation: + // this is what EndBackup's deferred forgetBackupSession does. + proposer.onPropose = nil + srv.forgetBackupSession(tok.pinID) + } + proposer.mu.Unlock() + + _, err = srv.RenewBackup(context.Background(), &pb.RenewBackupRequest{PinToken: begin.GetPinToken()}) + require.Equal(t, codes.Unavailable, status.Code(err)) + require.Contains(t, proposer.subtypes(), backupSubtypeRelease, + "a renewal whose session vanished must still release what it half-renewed") + require.Contains(t, proposer.subtypes(), backupSubtypeUnreserve, + "a renewal whose session vanished must still unreserve what it half-renewed") +} + +// backupProposalGroupError re-stamps any ResourceExhausted status, so a +// forwarded or proxied failure is indistinguishable from a real capacity +// rejection by code alone. Only kv.ErrTooManyActiveBackups as a Go error +// proves nothing was reserved; a bare status is ambiguous and must still be +// compensated, while keeping the client-facing code the caller expects. +func TestBeginBackupUnreservesAmbiguousResourceExhausted(t *testing.T) { + t.Parallel() + + group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000} + proposer := newBackupTestProposer() + proposer.failures[backupSubtypeReserve] = 8 + proposer.transportError[backupSubtypeReserve] = status.Error(codes.ResourceExhausted, "upstream quota exceeded") + srv := newBackupControlTestServer( + t, + &backupTestStore{}, + map[uint64]*backupTestGroup{1: group}, + map[uint64]*backupTestProposer{1: proposer}, + nil, + ) + + _, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{}) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + require.Contains(t, proposer.subtypes(), backupSubtypeUnreserve, + "an ambiguous ResourceExhausted must be compensated, not assumed definitive") +} + +// The definitive case must keep skipping compensation: kv.ErrTooManyActiveBackups +// on the apply response proves the reservation was refused, so an unreserve +// would be pure noise. +func TestBeginBackupSkipsUnreserveOnDefinitiveCapacityRejection(t *testing.T) { + t.Parallel() + + group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000} + proposer := newBackupTestProposer() + proposer.responseError[backupSubtypeReserve] = kv.ErrTooManyActiveBackups + srv := newBackupControlTestServer( + t, + &backupTestStore{}, + map[uint64]*backupTestGroup{1: group}, + map[uint64]*backupTestProposer{1: proposer}, + nil, + ) + + _, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{}) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + require.NotContains(t, proposer.subtypes(), backupSubtypeUnreserve, + "a definitive capacity rejection reserved nothing") +}