fix: bound component shutdown against unreachable peers (SYN-110)

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 {

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) {
@@ -215,10 +221,19 @@ 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(closingCtx, true)
+	if err != nil {
 		return false, err
 	}
-	_, curState := e.setClosing(true)
 	if curState == entryStateClosing {
 		ok = true
 		err = e.value.Close()
@@ -263,7 +278,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 +372,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 +411,14 @@ func (c *oCache) Close() (err error) {
 		toClose = append(toClose, e)
 	}
 	c.mu.Unlock()
+	// 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(context.Background(), 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)
 		}
 	}

app/ocache/ocache_test.go

@@ -748,3 +748,65 @@ 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")
+	}
+}
+
+// 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")
+}

commonspace/deletionmanager/deleteloop.go

@@ -2,17 +2,25 @@ package deletionmanager
 
 import (
 	"context"
+	"sync/atomic"
 	"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
 	deleteCancel context.CancelFunc
 	deleteChan   chan struct{}
 	deleteFunc   func(ctx context.Context)
 	loopDone     chan struct{}
+	closeTimeout time.Duration
+	running      atomic.Bool
 }
 
 func newDeleteLoop(deleteFunc func(ctx context.Context)) *deleteLoop {
@@ -23,10 +31,12 @@ func newDeleteLoop(deleteFunc func(ctx context.Context)) *deleteLoop {
 		deleteChan:   make(chan struct{}, 1),
 		deleteFunc:   deleteFunc,
 		loopDone:     make(chan struct{}),
+		closeTimeout: deleteCloseTimeout,
 	}
 }
 
 func (dl *deleteLoop) Run() {
+	dl.running.Store(true)
 	go dl.loop()
 }
 
@@ -55,7 +65,23 @@ 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
+	// 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 {
+	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")
+	}
 }

commonspace/deletionmanager/deleteloop_test.go

@@ -0,0 +1,95 @@
+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)
+}
+
+// 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")
+	}
+}

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

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
 }