From 8496abbce9689220e50e475fc485f2b5bb72924d Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 28 Jul 2026 17:28:32 +0200 Subject: [PATCH 1/5] fix(deletionmanager): make delete path ctx-cancellable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit deleteFunc reached two detached contexts — MarkTreeDeleted and the settings object AddContent broadcast — so a delete in flight against unreachable nodes never returned and deleteLoop.Close blocked until the app.Close watchdog panicked. Plumb the delete ctx through both, bail out of the queued/bound-child loops once it is cancelled, and bound the Close wait with a timeout and warning so a treemanager that ignores ctx can no longer wedge shutdown. --- commonspace/deletionmanager/deleteloop.go | 24 +++++- .../deletionmanager/deleteloop_test.go | 80 +++++++++++++++++++ commonspace/deletionmanager/deleter.go | 8 +- .../deletionmanager/deletionmanager.go | 2 +- commonspace/settings/settingsobject.go | 6 +- 5 files changed, 112 insertions(+), 8 deletions(-) create mode 100644 commonspace/deletionmanager/deleteloop_test.go diff --git a/commonspace/deletionmanager/deleteloop.go b/commonspace/deletionmanager/deleteloop.go index e7e1e1802..f2ac239f1 100644 --- a/commonspace/deletionmanager/deleteloop.go +++ b/commonspace/deletionmanager/deleteloop.go @@ -3,9 +3,14 @@ package deletionmanager import ( "context" "time" + + "go.uber.org/zap" ) -const deleteLoopInterval = time.Second * 20 +const ( + deleteLoopInterval = time.Second * 20 + deleteCloseTimeout = time.Second * 10 +) type deleteLoop struct { deleteCtx context.Context @@ -13,6 +18,7 @@ type deleteLoop struct { deleteChan chan struct{} deleteFunc func(ctx context.Context) loopDone chan struct{} + closeTimeout time.Duration } func newDeleteLoop(deleteFunc func(ctx context.Context)) *deleteLoop { @@ -23,6 +29,7 @@ func newDeleteLoop(deleteFunc func(ctx context.Context)) *deleteLoop { deleteChan: make(chan struct{}, 1), deleteFunc: deleteFunc, loopDone: make(chan struct{}), + closeTimeout: deleteCloseTimeout, } } @@ -55,7 +62,18 @@ func (dl *deleteLoop) notify() { } } -func (dl *deleteLoop) Close() { +// Close cancels the delete context and waits for the loop to exit. The wait is bounded: +// deleteFunc calls into treemanager implementations that may not honour ctx, and blocking +// here forever wedges the whole app.Close. +func (dl *deleteLoop) Close(ctx context.Context) { dl.deleteCancel() - <-dl.loopDone + timer := time.NewTimer(dl.closeTimeout) + defer timer.Stop() + select { + case <-dl.loopDone: + case <-ctx.Done(): + log.WarnCtx(ctx, "delete loop close interrupted, delete is still in flight", zap.Error(ctx.Err())) + case <-timer.C: + log.WarnCtx(ctx, "delete loop close timed out, delete is still in flight") + } } diff --git a/commonspace/deletionmanager/deleteloop_test.go b/commonspace/deletionmanager/deleteloop_test.go new file mode 100644 index 000000000..949ead12c --- /dev/null +++ b/commonspace/deletionmanager/deleteloop_test.go @@ -0,0 +1,80 @@ +package deletionmanager + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDeleteLoop_CloseCancelsDeleteCtx(t *testing.T) { + var ( + started = make(chan struct{}) + done = make(chan struct{}) + ) + dl := newDeleteLoop(func(ctx context.Context) { + close(started) + <-ctx.Done() + close(done) + }) + dl.Run() + <-started + dl.Close(context.Background()) + select { + case <-done: + default: + require.Fail(t, "deleteFunc was not cancelled") + } +} + +func TestDeleteLoop_CloseBoundedWhenDeleteFuncIgnoresCtx(t *testing.T) { + var ( + started = make(chan struct{}) + release = make(chan struct{}) + ) + dl := newDeleteLoop(func(ctx context.Context) { + close(started) + <-release + }) + dl.closeTimeout = 10 * time.Millisecond + dl.Run() + <-started + closed := make(chan struct{}) + go func() { + dl.Close(context.Background()) + close(closed) + }() + select { + case <-closed: + case <-time.After(time.Second): + require.Fail(t, "Close blocked on a deleteFunc that ignores ctx") + } + close(release) +} + +func TestDeleteLoop_CloseHonoursCtx(t *testing.T) { + var ( + started = make(chan struct{}) + release = make(chan struct{}) + ) + dl := newDeleteLoop(func(ctx context.Context) { + close(started) + <-release + }) + dl.Run() + <-started + ctx, cancel := context.WithCancel(context.Background()) + closed := make(chan struct{}) + go func() { + dl.Close(ctx) + close(closed) + }() + cancel() + select { + case <-closed: + case <-time.After(time.Second): + require.Fail(t, "Close ignored the cancelled ctx") + } + close(release) +} diff --git a/commonspace/deletionmanager/deleter.go b/commonspace/deletionmanager/deleter.go index ada11d988..a33490e68 100644 --- a/commonspace/deletionmanager/deleter.go +++ b/commonspace/deletionmanager/deleter.go @@ -36,6 +36,9 @@ func (d *deleter) Delete(ctx context.Context) { spaceId = d.st.Id() ) for _, id := range allQueued { + if ctx.Err() != nil { + return + } log := d.log.With(zap.String("treeId", id)) shouldDelete, err := d.tryMarkDeleted(ctx, spaceId, id) if !shouldDelete { @@ -67,6 +70,9 @@ func (d *deleter) deleteBoundChildren(ctx context.Context, spaceId, parentId str return } for _, child := range children { + if ctx.Err() != nil { + return + } if child.DeletedStatus >= headstorage.DeletedStatusDeleted { continue } @@ -100,5 +106,5 @@ func (d *deleter) tryMarkDeleted(ctx context.Context, spaceId, treeId string) (b if !errors.Is(err, treestorage.ErrUnknownTreeId) { return false, err } - return false, d.getter.MarkTreeDeleted(context.Background(), spaceId, treeId) + return false, d.getter.MarkTreeDeleted(ctx, spaceId, treeId) } diff --git a/commonspace/deletionmanager/deletionmanager.go b/commonspace/deletionmanager/deletionmanager.go index c490a35fa..8981c1cbd 100644 --- a/commonspace/deletionmanager/deletionmanager.go +++ b/commonspace/deletionmanager/deletionmanager.go @@ -61,7 +61,7 @@ func (d *deletionManager) Run(ctx context.Context) (err error) { } func (d *deletionManager) Close(ctx context.Context) (err error) { - d.loop.Close() + d.loop.Close(ctx) return } diff --git a/commonspace/settings/settingsobject.go b/commonspace/settings/settingsobject.go index 9288c207a..0cc4b9aea 100644 --- a/commonspace/settings/settingsobject.go +++ b/commonspace/settings/settingsobject.go @@ -230,12 +230,12 @@ func (s *settingsObject) DeleteObject(ctx context.Context, id string) (err error return } - return s.addContent(res, isSnapshot) + return s.addContent(ctx, res, isSnapshot) } -func (s *settingsObject) addContent(data []byte, isSnapshot bool) (err error) { +func (s *settingsObject) addContent(ctx context.Context, data []byte, isSnapshot bool) (err error) { accountData := s.account.Account() - res, err := s.AddContent(context.Background(), objecttree.SignableChangeContent{ + res, err := s.AddContent(ctx, objecttree.SignableChangeContent{ Data: data, Key: accountData.SignKey, IsSnapshot: isSnapshot, From 2328e3413cabb52765c70cbfd8c53f3b28955c80 Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 28 Jul 2026 17:47:03 +0200 Subject: [PATCH 2/5] fix(keyvalue,syncacl): bound shutdown on in-flight peer rpc Two more instances of the deletionmanager hang. concurrentLimiter.Close waited on the WaitGroup with no bound while the scheduled goroutine was inside a peer sync that only checks ctx before it starts. syncAcl broadcast under context.Background while holding the acl lock, which Close then waits for, so an acl record arriving against unreachable nodes wedged Close forever. Bound the limiter wait with a timeout and the close ctx, give syncAcl a cancellable broadcast ctx, and cancel it before Close takes the lock. --- commonspace/object/acl/syncacl/syncacl.go | 11 +++- commonspace/object/keyvalue/keyvalue.go | 2 +- .../object/keyvalue/keyvalue_ordering_test.go | 6 +- commonspace/object/keyvalue/keyvalue_test.go | 8 +-- commonspace/object/keyvalue/limiter.go | 35 +++++++++-- commonspace/object/keyvalue/limiter_test.go | 60 +++++++++++++++++++ 6 files changed, 107 insertions(+), 15 deletions(-) create mode 100644 commonspace/object/keyvalue/limiter_test.go diff --git a/commonspace/object/acl/syncacl/syncacl.go b/commonspace/object/acl/syncacl/syncacl.go index 738991890..e9de38d69 100644 --- a/commonspace/object/acl/syncacl/syncacl.go +++ b/commonspace/object/acl/syncacl/syncacl.go @@ -48,6 +48,10 @@ type syncAcl struct { verifier recordverifier.RecordVerifier isClosed bool aclUpdater headupdater.AclUpdater + // broadcasts happen under the acl lock, so they get their own cancellable ctx: + // Close cancels it before taking the lock + broadcastCtx context.Context + broadcastCancel context.CancelFunc } func (s *syncAcl) SetAclUpdater(updater headupdater.AclUpdater) { @@ -63,6 +67,7 @@ func (s *syncAcl) Run(ctx context.Context) (err error) { } func (s *syncAcl) Init(a *app.App) (err error) { + s.broadcastCtx, s.broadcastCancel = context.WithCancel(context.Background()) storage := a.MustComponent(spacestorage.CName).(spacestorage.SpaceStorage) aclStorage, err := storage.AclStorage() if err != nil { @@ -103,7 +108,7 @@ func (s *syncAcl) AddRawRecord(rawRec *consensusproto.RawRecordWithId) (err erro } func (s *syncAcl) broadcast(headUpdate *objectmessages.HeadUpdate) { - err := s.syncClient.Broadcast(context.Background(), headUpdate) + err := s.syncClient.Broadcast(s.broadcastCtx, headUpdate) if err != nil { log.Error("broadcast acl message error", zap.Error(err)) } @@ -139,6 +144,10 @@ func (s *syncAcl) SyncWithPeer(ctx context.Context, p peer.Peer) (err error) { } func (s *syncAcl) Close(ctx context.Context) (err error) { + // unblocks a broadcast holding the acl lock, otherwise the Lock below never returns + if s.broadcastCancel != nil { + s.broadcastCancel() + } if s.AclList == nil { return } diff --git a/commonspace/object/keyvalue/keyvalue.go b/commonspace/object/keyvalue/keyvalue.go index c634d2a36..01ca32594 100644 --- a/commonspace/object/keyvalue/keyvalue.go +++ b/commonspace/object/keyvalue/keyvalue.go @@ -274,7 +274,7 @@ func (k *keyValueService) Run(ctx context.Context) (err error) { func (k *keyValueService) Close(ctx context.Context) (err error) { k.cancel() - k.limiter.Close() + k.limiter.Close(ctx) return nil } diff --git a/commonspace/object/keyvalue/keyvalue_ordering_test.go b/commonspace/object/keyvalue/keyvalue_ordering_test.go index 4b8ac0620..ccceeeee1 100644 --- a/commonspace/object/keyvalue/keyvalue_ordering_test.go +++ b/commonspace/object/keyvalue/keyvalue_ordering_test.go @@ -47,7 +47,7 @@ func TestStoreElementsNewestFirst(t *testing.T) { } require.NoError(t, fxClient.SyncWithPeer(serverPeer)) - fxClient.limiter.Close() + fxClient.limiter.Close(ctx) var sentKeys []string for _, id := range fxServer.ts.sentIds() { @@ -74,7 +74,7 @@ func TestPushedValuesPersistDespiteSendFailure(t *testing.T) { fxServer.ts.setFailTerminator(true) require.NoError(t, fxClient.SyncWithPeer(serverPeer)) - fxClient.limiter.Close() + fxClient.limiter.Close(ctx) require.True(t, fxServer.check(t, "pushed", []byte("pushed-value")), "server must persist pushed values even when its response send fails") @@ -91,7 +91,7 @@ func TestIncrementalApplyConvergence(t *testing.T) { } require.NoError(t, fxClient.SyncWithPeer(serverPeer)) - fxClient.limiter.Close() + fxClient.limiter.Close(ctx) // The client broadcasts once per applied SetRaw, so the broadcast count // proves the pull was applied incrementally rather than in one shot. diff --git a/commonspace/object/keyvalue/keyvalue_test.go b/commonspace/object/keyvalue/keyvalue_test.go index 41b654012..23ebf0f1c 100644 --- a/commonspace/object/keyvalue/keyvalue_test.go +++ b/commonspace/object/keyvalue/keyvalue_test.go @@ -40,7 +40,7 @@ func TestKeyValueService(t *testing.T) { fxServer.add(t, "key4", []byte("value4")) err := fxClient.SyncWithPeer(serverPeer) require.NoError(t, err) - fxClient.limiter.Close() + fxClient.limiter.Close(ctx) fxClient.check(t, "key3", []byte("value3")) fxClient.check(t, "key4", []byte("value4")) fxServer.check(t, "key1", []byte("value1")) @@ -53,7 +53,7 @@ func TestKeyValueService(t *testing.T) { fxServer.add(t, "key1", []byte("value2")) err := fxClient.SyncWithPeer(serverPeer) require.NoError(t, err) - fxClient.limiter.Close() + fxClient.limiter.Close(ctx) fxClient.check(t, "key1", []byte("value1")) fxClient.check(t, "key1", []byte("value2")) fxServer.check(t, "key1", []byte("value1")) @@ -62,7 +62,7 @@ func TestKeyValueService(t *testing.T) { fxServer.add(t, "key1", []byte("value2-2")) err = fxClient.SyncWithPeer(serverPeer) require.NoError(t, err) - fxClient.limiter.Close() + fxClient.limiter.Close(ctx) fxClient.check(t, "key1", []byte("value1-2")) fxClient.check(t, "key1", []byte("value2-2")) fxServer.check(t, "key1", []byte("value1-2")) @@ -100,7 +100,7 @@ func TestKeyValueService(t *testing.T) { } err := fxClient.SyncWithPeer(serverPeer) require.NoError(t, err) - fxClient.limiter.Close() + fxClient.limiter.Close(ctx) for key := range allKeys { if strings.HasPrefix(key, "client-key-") { diff --git a/commonspace/object/keyvalue/limiter.go b/commonspace/object/keyvalue/limiter.go index 7a36bffdc..9f554d871 100644 --- a/commonspace/object/keyvalue/limiter.go +++ b/commonspace/object/keyvalue/limiter.go @@ -3,17 +3,24 @@ package keyvalue import ( "context" "sync" + "time" + + "go.uber.org/zap" ) +const limiterCloseTimeout = time.Second * 10 + type concurrentLimiter struct { - mu sync.Mutex - inProgress map[string]bool - wg sync.WaitGroup + mu sync.Mutex + inProgress map[string]bool + wg sync.WaitGroup + closeTimeout time.Duration } func newConcurrentLimiter() *concurrentLimiter { return &concurrentLimiter{ - inProgress: make(map[string]bool), + inProgress: make(map[string]bool), + closeTimeout: limiterCloseTimeout, } } @@ -47,6 +54,22 @@ func (cl *concurrentLimiter) ScheduleRequest(ctx context.Context, id string, act return true } -func (cl *concurrentLimiter) Close() { - cl.wg.Wait() +// Close waits for the scheduled requests to finish. The wait is bounded: a request +// already past the ctx check is doing peer rpc that may not return while nodes are +// unreachable, and blocking here forever wedges the whole app.Close. +func (cl *concurrentLimiter) Close(ctx context.Context) { + done := make(chan struct{}) + go func() { + cl.wg.Wait() + close(done) + }() + timer := time.NewTimer(cl.closeTimeout) + defer timer.Stop() + select { + case <-done: + case <-ctx.Done(): + log.WarnCtx(ctx, "key value close interrupted, peer sync is still in flight", zap.Error(ctx.Err())) + case <-timer.C: + log.WarnCtx(ctx, "key value close timed out, peer sync is still in flight") + } } diff --git a/commonspace/object/keyvalue/limiter_test.go b/commonspace/object/keyvalue/limiter_test.go new file mode 100644 index 000000000..4ae7812ea --- /dev/null +++ b/commonspace/object/keyvalue/limiter_test.go @@ -0,0 +1,60 @@ +package keyvalue + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestConcurrentLimiter_CloseBoundedWhenActionIgnoresCtx(t *testing.T) { + var ( + started = make(chan struct{}) + release = make(chan struct{}) + ) + cl := newConcurrentLimiter() + cl.closeTimeout = 10 * time.Millisecond + require.True(t, cl.ScheduleRequest(context.Background(), "peer", func() { + close(started) + <-release + })) + <-started + closed := make(chan struct{}) + go func() { + cl.Close(context.Background()) + close(closed) + }() + select { + case <-closed: + case <-time.After(time.Second): + require.Fail(t, "Close blocked on a request that ignores ctx") + } + close(release) +} + +func TestConcurrentLimiter_CloseHonoursCtx(t *testing.T) { + var ( + started = make(chan struct{}) + release = make(chan struct{}) + ) + cl := newConcurrentLimiter() + require.True(t, cl.ScheduleRequest(context.Background(), "peer", func() { + close(started) + <-release + })) + <-started + ctx, cancel := context.WithCancel(context.Background()) + closed := make(chan struct{}) + go func() { + cl.Close(ctx) + close(closed) + }() + cancel() + select { + case <-closed: + case <-time.After(time.Second): + require.Fail(t, "Close ignored the cancelled ctx") + } + close(release) +} From fd0a7b0858234655adcc6a409b8563cd2d19add6 Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 28 Jul 2026 17:49:18 +0200 Subject: [PATCH 3/5] fix: cancel queued sync actions on close, survive Close before Run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit actionPool created a cancel func and never called it, so every queued action ran under an effectively uncancellable ctx and kept issuing peer requests after close — it is also the ctx a syncTree holds its lock under, which made settings.Close unrecoverable. app.Start's closeServices closes components whose Run never executed, which deadlocked subscribeClient.Close on a channel only streamWatcher closes and nil-dereferenced actionPool.periodicLoop and streamPool.dial. Guard all three. --- coordinator/subscribeclient/client.go | 11 +++++++- coordinator/subscribeclient/client_test.go | 32 ++++++++++++++++++++++ net/streampool/closewithoutrun_test.go | 20 ++++++++++++++ net/streampool/streampool.go | 5 +++- util/syncqueues/actionpool.go | 7 ++++- util/syncqueues/actionpool_test.go | 30 ++++++++++++++++++++ 6 files changed, 102 insertions(+), 3 deletions(-) create mode 100644 coordinator/subscribeclient/client_test.go create mode 100644 net/streampool/closewithoutrun_test.go diff --git a/coordinator/subscribeclient/client.go b/coordinator/subscribeclient/client.go index 289067a8e..b72080f3c 100644 --- a/coordinator/subscribeclient/client.go +++ b/coordinator/subscribeclient/client.go @@ -43,6 +43,7 @@ type subscribeClient struct { ctx context.Context ctxCancel context.CancelFunc close chan struct{} + running bool } func (s *subscribeClient) Init(a *app.App) (err error) { @@ -60,18 +61,26 @@ func (s *subscribeClient) Name() (name string) { } func (s *subscribeClient) Run(ctx context.Context) error { + s.mu.Lock() + s.running = true + s.mu.Unlock() go s.streamWatcher() return nil } func (s *subscribeClient) Close(_ context.Context) (err error) { s.mu.Lock() + running := s.running if s.stream != nil { _ = s.stream.Close() } s.mu.Unlock() s.ctxCancel() - <-s.close + // s.close is closed by streamWatcher, which only exists after Run: app.Start + // closes components it never ran + if running { + <-s.close + } return nil } diff --git a/coordinator/subscribeclient/client_test.go b/coordinator/subscribeclient/client_test.go new file mode 100644 index 000000000..ecbec0d6d --- /dev/null +++ b/coordinator/subscribeclient/client_test.go @@ -0,0 +1,32 @@ +package subscribeclient + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/anyproto/any-sync/coordinator/coordinatorproto" +) + +// app.Start closes components it never ran: a failing Init makes closeServices +// call Close on every component registered before it. s.close is only closed by +// streamWatcher, which Run starts. +func TestSubscribeClient_CloseWithoutRun(t *testing.T) { + s := &subscribeClient{} + s.ctx, s.ctxCancel = context.WithCancel(context.Background()) + s.close = make(chan struct{}) + s.callbacks = make(map[coordinatorproto.NotifyEventType]EventCallback) + + closed := make(chan struct{}) + go func() { + require.NoError(t, s.Close(context.Background())) + close(closed) + }() + select { + case <-closed: + case <-time.After(time.Second): + require.Fail(t, "Close deadlocked when Run was never called") + } +} diff --git a/net/streampool/closewithoutrun_test.go b/net/streampool/closewithoutrun_test.go new file mode 100644 index 000000000..d5f1ae2c3 --- /dev/null +++ b/net/streampool/closewithoutrun_test.go @@ -0,0 +1,20 @@ +package streampool + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/anyproto/any-sync/app/debugstat" +) + +// app.Start closes components it never ran: a failing Init makes closeServices +// call Close on every component registered before it. +func TestStreamPool_CloseWithoutRun(t *testing.T) { + s := New().(*streamPool) + s.statService = debugstat.NewNoOp() + require.NotPanics(t, func() { + require.NoError(t, s.Close(context.Background())) + }) +} diff --git a/net/streampool/streampool.go b/net/streampool/streampool.go index d7a72c843..9c5a2d333 100644 --- a/net/streampool/streampool.go +++ b/net/streampool/streampool.go @@ -450,7 +450,10 @@ func (s *streamPool) Close(ctx context.Context) (err error) { if s.metric != nil { s.metric.UnregisterStreamPoolSyncMetric() } - return s.dial.Close() + if s.dial != nil { + return s.dial.Close() + } + return nil } func removeStream(m map[string][]uint32, key string, streamId uint32) { diff --git a/util/syncqueues/actionpool.go b/util/syncqueues/actionpool.go index e80a204a8..bca25ae75 100644 --- a/util/syncqueues/actionpool.go +++ b/util/syncqueues/actionpool.go @@ -109,7 +109,12 @@ func (rp *actionPool) Add(peerId, objectId string, action func(ctx context.Conte } func (rp *actionPool) Close() { - rp.periodicLoop.Close() + // cancels the ctx handed to every queued action, otherwise in-flight peer + // requests keep running after close + rp.cancel() + if rp.periodicLoop != nil { + rp.periodicLoop.Close() + } rp.mu.Lock() defer rp.mu.Unlock() rp.isClosed = true diff --git a/util/syncqueues/actionpool_test.go b/util/syncqueues/actionpool_test.go index 81b482f13..434159789 100644 --- a/util/syncqueues/actionpool_test.go +++ b/util/syncqueues/actionpool_test.go @@ -152,3 +152,33 @@ func TestRequestPool(t *testing.T) { rp.Close() }) } + +// app.Start closes components it never ran: a failing Init makes closeServices +// call Close on every component registered before it. +func TestActionPool_CloseWithoutRun(t *testing.T) { + rp := NewActionPool(time.Minute, time.Minute, func(peerId string) *replaceableQueue { + return newReplaceableQueue(1, 1) + }) + require.NotPanics(t, rp.Close) +} + +func TestActionPool_CloseCancelsActionCtx(t *testing.T) { + rp := NewActionPool(time.Minute, time.Minute, func(peerId string) *replaceableQueue { + return newReplaceableQueue(1, 1) + }) + rp.Run() + started := make(chan struct{}) + cancelled := make(chan struct{}) + rp.Add("peerId", "objectId", func(ctx context.Context) { + close(started) + <-ctx.Done() + close(cancelled) + }, func() {}) + <-started + rp.Close() + select { + case <-cancelled: + case <-time.After(time.Second): + require.Fail(t, "in-flight action kept running after Close") + } +} From 7f4a8177d4973688022f2a512b9a64fb3c2a3c29 Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 28 Jul 2026 17:56:28 +0200 Subject: [PATCH 4/5] fix(net,ocache): bound close against unresponsive peers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ocache.Close waited with context.Background on entries the gc was already closing, and that gc sits in peer.TryClose -> drpc conn Close -> a yamux stream whose peer never sends FIN, which yamux bounds at 5 minutes — past the app close deadline. Give Close one deadline for the whole pass; an entry still held by another closer is left to it instead of blocking shutdown. peer.gc closed those conns while holding p.mu, freezing AcquireDrpcConn for the same peer; close them after the lock is released. Set the yamux stream close timeout from the write timeout, and bound the webtransport dial with DialTimeoutSec, which until now applied only to the accept path. --- app/ocache/entry.go | 14 +++++++-- app/ocache/ocache.go | 22 +++++++++++--- app/ocache/ocache_test.go | 30 +++++++++++++++++++ net/peer/peer.go | 13 ++++++-- .../webtransport/webtransport_native.go | 4 +++ net/transport/yamux/yamux.go | 3 ++ 6 files changed, 78 insertions(+), 8 deletions(-) diff --git a/app/ocache/entry.go b/app/ocache/entry.go index 87b20db59..8c47c155d 100644 --- a/app/ocache/entry.go +++ b/app/ocache/entry.go @@ -97,7 +97,10 @@ func (e *entry) waitClose(ctx context.Context, id string) (res bool, err error) } } -func (e *entry) setClosing(wait bool) (prevState, curState entryState) { +// setClosing transitions the entry to closing. With wait it blocks until another +// closer is done with it, bounded by ctx: that closer may be inside a TryClose +// that waits on an unresponsive peer. +func (e *entry) setClosing(ctx context.Context, wait bool) (prevState, curState entryState, err error) { e.mx.Lock() prevState = e.state curState = e.state @@ -112,7 +115,14 @@ func (e *entry) setClosing(wait bool) (prevState, curState entryState) { if !wait { return } - <-waitCh + select { + case <-waitCh: + case <-ctx.Done(): + e.mx.Lock() + curState = e.state + e.mx.Unlock() + return prevState, curState, ctx.Err() + } e.mx.Lock() } if e.state != entryStateClosed { diff --git a/app/ocache/ocache.go b/app/ocache/ocache.go index 7c63ec58f..a301702df 100644 --- a/app/ocache/ocache.go +++ b/app/ocache/ocache.go @@ -21,6 +21,8 @@ var ( var ( defaultTTL = time.Minute defaultGC = 20 * time.Second + // bounds Close against entries the gc is already closing + closeTimeout = 10 * time.Second ) var log = logger.NewNamed("ocache") @@ -56,6 +58,8 @@ func New(loadFunc LoadFunc, opts ...Option) OCache { gc: defaultGC, closeCh: make(chan struct{}), log: log.Sugar(), + + closeTimeout: closeTimeout, } for _, o := range opts { if o != nil { @@ -119,6 +123,8 @@ type oCache struct { closeCh chan struct{} log *zap.SugaredLogger metrics *metrics + + closeTimeout time.Duration } func (c *oCache) Get(ctx context.Context, id string) (value Object, err error) { @@ -218,7 +224,10 @@ func (c *oCache) remove(ctx context.Context, e *entry) (ok bool, err error) { if _, err = e.waitLoad(ctx, e.id); err != nil { return false, err } - _, curState := e.setClosing(true) + _, curState, err := e.setClosing(ctx, true) + if err != nil { + return false, err + } if curState == entryStateClosing { ok = true err = e.value.Close() @@ -263,7 +272,7 @@ func (c *oCache) TryRemove(id string) (ok bool, err error) { c.mu.Unlock() - prevState, _ := e.setClosing(false) + prevState, _, _ := e.setClosing(context.Background(), false) if prevState == entryStateClosing || prevState == entryStateClosed { return false, nil } @@ -357,7 +366,7 @@ func (c *oCache) GC() { c.mu.Unlock() closedNum := 0 for _, e := range toClose { - prevState, _ := e.setClosing(false) + prevState, _, _ := e.setClosing(context.Background(), false) if prevState == entryStateClosing || prevState == entryStateClosed { continue } @@ -396,8 +405,13 @@ func (c *oCache) Close() (err error) { toClose = append(toClose, e) } c.mu.Unlock() + // one deadline for the whole close: an entry being closed by the gc can be + // stuck in TryClose on an unresponsive peer, and waiting per entry would let + // the total grow with the cache size + ctx, cancel := context.WithTimeout(context.Background(), c.closeTimeout) + defer cancel() for _, e := range toClose { - if _, err := c.remove(context.Background(), e); err != nil && err != ErrNotExists { + if _, err := c.remove(ctx, e); err != nil && err != ErrNotExists { c.log.With("object_id", e.id).Warnf("cache close: object close error: %v", err) } } diff --git a/app/ocache/ocache_test.go b/app/ocache/ocache_test.go index 4d3ed7917..4cee1b3b3 100644 --- a/app/ocache/ocache_test.go +++ b/app/ocache/ocache_test.go @@ -748,3 +748,33 @@ func TestOCache_RemoveBusyRevertDoubleClose(t *testing.T) { } } } + +// An entry another closer already owns must not stall cache close: that closer +// can be a gc sitting in TryClose against an unresponsive peer. +func TestOCache_CloseBoundedByOtherCloser(t *testing.T) { + c := New(func(ctx context.Context, id string) (Object, error) { + return NewTestObject(id, true, nil), nil + }) + oc := c.(*oCache) + oc.closeTimeout = 20 * time.Millisecond + _, err := c.Get(ctx, "id") + require.NoError(t, err) + + oc.mu.Lock() + e := oc.data["id"] + oc.mu.Unlock() + _, curState, err := e.setClosing(ctx, false) + require.NoError(t, err) + require.Equal(t, entryState(entryStateClosing), curState) + + done := make(chan struct{}) + go func() { + _ = c.Close() + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + require.Fail(t, "Close blocked behind an entry owned by another closer") + } +} diff --git a/net/peer/peer.go b/net/peer/peer.go index 56d193177..100e1c4d3 100644 --- a/net/peer/peer.go +++ b/net/peer/peer.go @@ -380,6 +380,15 @@ func (p *peer) TryClose(objectTTL time.Duration) (res bool, err error) { } func (p *peer) gc(ttl time.Duration) (aliveCount int) { + // drpc conn Close blocks until its reader unwinds, which on a stalled stream + // takes until the yamux stream close timeout: collect the doomed conns and + // close them after releasing the lock + var toClose []*subConn + defer func() { + for _, conn := range toClose { + _ = conn.Close() + } + }() p.mu.Lock() defer p.mu.Unlock() minLastUsage := time.Now().Add(-ttl) @@ -392,7 +401,7 @@ func (p *peer) gc(ttl time.Duration) (aliveCount int) { default: } if in.LastUsage().Before(minLastUsage) { - _ = in.Close() + toClose = append(toClose, in) p.inactive[i] = nil hasClosed = true } @@ -415,7 +424,7 @@ func (p *peer) gc(ttl time.Duration) (aliveCount int) { } if act.LastUsage().Before(minLastUsage) { log.Warn("close active connection because no activity", zap.String("peerId", p.id), zap.String("addr", p.Addr())) - _ = act.Close() + toClose = append(toClose, act) delete(p.active, act) continue } diff --git a/net/transport/webtransport/webtransport_native.go b/net/transport/webtransport/webtransport_native.go index 8573e7e3c..5ca23266a 100644 --- a/net/transport/webtransport/webtransport_native.go +++ b/net/transport/webtransport/webtransport_native.go @@ -206,6 +206,10 @@ func (t *wtTransport) Dial(ctx context.Context, addr string) (transport.MultiCon if expectedPeerId == "" { return nil, fmt.Errorf("no expected peer id in context for WebTransport dial") } + // a peer that completes the quic handshake and then stalls keeps the idle + // timeout from firing, so bound the dial like yamux does + ctx, cancel := context.WithTimeout(ctx, time.Duration(t.conf.DialTimeoutSec)*time.Second) + defer cancel() dialer := wt.Dialer{ TLSClientConfig: &tls.Config{ diff --git a/net/transport/yamux/yamux.go b/net/transport/yamux/yamux.go index 0530ca8be..ef921ad7a 100644 --- a/net/transport/yamux/yamux.go +++ b/net/transport/yamux/yamux.go @@ -63,6 +63,9 @@ func (y *yamuxTransport) Init(a *app.App) (err error) { } } y.yamuxConf.StreamOpenTimeout = time.Duration(y.conf.DialTimeoutSec) * time.Second + // yamux defaults to 5 minutes, longer than the app close deadline: a stream + // whose peer never sends FIN back would keep drpc conn Close blocked + y.yamuxConf.StreamCloseTimeout = time.Duration(y.conf.WriteTimeoutSec) * time.Second y.yamuxConf.ConnectionWriteTimeout = time.Duration(y.conf.WriteTimeoutSec) * time.Second y.listCtx, y.listCtxCancel = context.WithCancel(context.Background()) return From d1c41ba84f714ae6b1b6dbb561a6c62048c0420b Mon Sep 17 00:00:00 2001 From: Sergey Cherepanov Date: Tue, 28 Jul 2026 18:28:52 +0200 Subject: [PATCH 5/5] fix: review fixes for the bounded close paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit concurrentLimiter.Close returning on timeout left a waiter parked on the WaitGroup; a later ScheduleRequest panicked with "WaitGroup is reused before previous Wait has returned", and SyncWithPeer stays callable after close. Reject requests once closed. ocache.Close passed its deadline to waitLoad too, where an expired ctx and a finished load are both ready and select picks at random — roughly half the healthy entries returned early and were never closed. Spend the deadline only on entries another closer holds. deleteLoop.Close burned the full timeout when Run never ran. --- app/ocache/ocache.go | 21 ++++++++---- app/ocache/ocache_test.go | 32 +++++++++++++++++++ commonspace/deletionmanager/deleteloop.go | 8 +++++ .../deletionmanager/deleteloop_test.go | 15 +++++++++ commonspace/object/keyvalue/limiter.go | 16 +++++++--- commonspace/object/keyvalue/limiter_test.go | 28 ++++++++++++++++ 6 files changed, 109 insertions(+), 11 deletions(-) diff --git a/app/ocache/ocache.go b/app/ocache/ocache.go index a301702df..5422247fb 100644 --- a/app/ocache/ocache.go +++ b/app/ocache/ocache.go @@ -221,10 +221,16 @@ func (c *oCache) closeAndDelete(e *entry) { } func (c *oCache) remove(ctx context.Context, e *entry) (ok bool, err error) { - if _, err = e.waitLoad(ctx, e.id); err != nil { + return c.removeCtx(ctx, ctx, e) +} + +// loadCtx bounds waiting for an in-flight load, closingCtx bounds waiting for +// another closer to release the entry +func (c *oCache) removeCtx(loadCtx, closingCtx context.Context, e *entry) (ok bool, err error) { + if _, err = e.waitLoad(loadCtx, e.id); err != nil { return false, err } - _, curState, err := e.setClosing(ctx, true) + _, curState, err := e.setClosing(closingCtx, true) if err != nil { return false, err } @@ -405,13 +411,14 @@ func (c *oCache) Close() (err error) { toClose = append(toClose, e) } c.mu.Unlock() - // one deadline for the whole close: an entry being closed by the gc can be - // stuck in TryClose on an unresponsive peer, and waiting per entry would let - // the total grow with the cache size - ctx, cancel := context.WithTimeout(context.Background(), c.closeTimeout) + // one deadline for the whole pass, spent only on entries another closer holds: + // that closer can be a gc stuck in TryClose on an unresponsive peer. Loads are + // already cancelled above, and value.Close takes no ctx, so uncontended entries + // still close normally once the deadline has passed. + closingCtx, cancel := context.WithTimeout(context.Background(), c.closeTimeout) defer cancel() for _, e := range toClose { - if _, err := c.remove(ctx, e); err != nil && err != ErrNotExists { + if _, err := c.removeCtx(context.Background(), closingCtx, e); err != nil && err != ErrNotExists { c.log.With("object_id", e.id).Warnf("cache close: object close error: %v", err) } } diff --git a/app/ocache/ocache_test.go b/app/ocache/ocache_test.go index 4cee1b3b3..f06ff2f0a 100644 --- a/app/ocache/ocache_test.go +++ b/app/ocache/ocache_test.go @@ -778,3 +778,35 @@ func TestOCache_CloseBoundedByOtherCloser(t *testing.T) { require.Fail(t, "Close blocked behind an entry owned by another closer") } } + +// The close deadline is spent only on entries another closer holds: healthy +// entries must still be closed after it has expired. +func TestOCache_CloseClosesHealthyEntriesAfterDeadline(t *testing.T) { + var objects []*testObject + c := New(func(ctx context.Context, id string) (Object, error) { + o := NewTestObject(id, true, nil) + objects = append(objects, o) + return o, nil + }) + oc := c.(*oCache) + oc.closeTimeout = time.Millisecond + for i := 0; i < 50; i++ { + _, err := c.Get(ctx, fmt.Sprint(i)) + require.NoError(t, err) + } + // one entry is held by another closer and will eat the whole deadline + oc.mu.Lock() + held := oc.data["0"] + oc.mu.Unlock() + _, _, err := held.setClosing(ctx, false) + require.NoError(t, err) + + require.NoError(t, c.Close()) + var notClosed []string + for _, o := range objects { + if o.name != "0" && !o.closeCalled { + notClosed = append(notClosed, o.name) + } + } + require.Empty(t, notClosed, "entries nobody else holds must still be closed after the deadline") +} diff --git a/commonspace/deletionmanager/deleteloop.go b/commonspace/deletionmanager/deleteloop.go index f2ac239f1..eafcde87e 100644 --- a/commonspace/deletionmanager/deleteloop.go +++ b/commonspace/deletionmanager/deleteloop.go @@ -2,6 +2,7 @@ package deletionmanager import ( "context" + "sync/atomic" "time" "go.uber.org/zap" @@ -19,6 +20,7 @@ type deleteLoop struct { deleteFunc func(ctx context.Context) loopDone chan struct{} closeTimeout time.Duration + running atomic.Bool } func newDeleteLoop(deleteFunc func(ctx context.Context)) *deleteLoop { @@ -34,6 +36,7 @@ func newDeleteLoop(deleteFunc func(ctx context.Context)) *deleteLoop { } func (dl *deleteLoop) Run() { + dl.running.Store(true) go dl.loop() } @@ -67,6 +70,11 @@ func (dl *deleteLoop) notify() { // here forever wedges the whole app.Close. func (dl *deleteLoop) Close(ctx context.Context) { dl.deleteCancel() + // loopDone is closed by loop, which only Run starts: app.Start closes + // components it never ran + if !dl.running.Load() { + return + } timer := time.NewTimer(dl.closeTimeout) defer timer.Stop() select { diff --git a/commonspace/deletionmanager/deleteloop_test.go b/commonspace/deletionmanager/deleteloop_test.go index 949ead12c..0018812b2 100644 --- a/commonspace/deletionmanager/deleteloop_test.go +++ b/commonspace/deletionmanager/deleteloop_test.go @@ -78,3 +78,18 @@ func TestDeleteLoop_CloseHonoursCtx(t *testing.T) { } close(release) } + +// app.Start closes components it never ran: loopDone is only closed by the loop. +func TestDeleteLoop_CloseWithoutRun(t *testing.T) { + dl := newDeleteLoop(func(ctx context.Context) {}) + closed := make(chan struct{}) + go func() { + dl.Close(context.Background()) + close(closed) + }() + select { + case <-closed: + case <-time.After(time.Second): + require.Fail(t, "Close waited for a loop that was never started") + } +} diff --git a/commonspace/object/keyvalue/limiter.go b/commonspace/object/keyvalue/limiter.go index 9f554d871..8b4ee9733 100644 --- a/commonspace/object/keyvalue/limiter.go +++ b/commonspace/object/keyvalue/limiter.go @@ -14,6 +14,7 @@ type concurrentLimiter struct { mu sync.Mutex inProgress map[string]bool wg sync.WaitGroup + closed bool closeTimeout time.Duration } @@ -26,7 +27,10 @@ func newConcurrentLimiter() *concurrentLimiter { func (cl *concurrentLimiter) ScheduleRequest(ctx context.Context, id string, action func()) bool { cl.mu.Lock() - if cl.inProgress[id] { + // a bounded Close can return while wg still has waiters parked: an Add after + // that panics with "WaitGroup is reused before previous Wait has returned", + // and SyncWithPeer stays callable after Close + if cl.closed || cl.inProgress[id] { cl.mu.Unlock() return false } @@ -54,10 +58,14 @@ func (cl *concurrentLimiter) ScheduleRequest(ctx context.Context, id string, act return true } -// Close waits for the scheduled requests to finish. The wait is bounded: a request -// already past the ctx check is doing peer rpc that may not return while nodes are -// unreachable, and blocking here forever wedges the whole app.Close. +// Close rejects further requests and waits for the scheduled ones to finish. The +// wait is bounded: a request already past the ctx check is doing peer rpc that may +// not return while nodes are unreachable, and blocking here forever wedges the +// whole app.Close. On timeout the request is abandoned, not stopped. func (cl *concurrentLimiter) Close(ctx context.Context) { + cl.mu.Lock() + cl.closed = true + cl.mu.Unlock() done := make(chan struct{}) go func() { cl.wg.Wait() diff --git a/commonspace/object/keyvalue/limiter_test.go b/commonspace/object/keyvalue/limiter_test.go index 4ae7812ea..592a428c3 100644 --- a/commonspace/object/keyvalue/limiter_test.go +++ b/commonspace/object/keyvalue/limiter_test.go @@ -2,6 +2,7 @@ package keyvalue import ( "context" + "sync" "testing" "time" @@ -58,3 +59,30 @@ func TestConcurrentLimiter_CloseHonoursCtx(t *testing.T) { } close(release) } + +// A timed-out Close leaves a waiter parked on the WaitGroup: a later Add would +// panic with "WaitGroup is reused before previous Wait has returned". +func TestConcurrentLimiter_NoScheduleAfterClose(t *testing.T) { + for i := 0; i < 500; i++ { + cl := newConcurrentLimiter() + cl.closeTimeout = time.Microsecond + release := make(chan struct{}) + require.True(t, cl.ScheduleRequest(context.Background(), "peer", func() { + <-release + })) + cl.Close(context.Background()) + + var start sync.WaitGroup + start.Add(2) + go func() { + start.Done() + start.Wait() + close(release) + }() + go func() { + start.Done() + start.Wait() + require.False(t, cl.ScheduleRequest(context.Background(), "peer2", func() {})) + }() + } +}