Merge pull request #2464 from anyproto/go-5748-fix-data-races-and-deadlocks

Go 5748 fix data races and deadlocks

core/filestorage/rpcstore/store.go

@@ -226,15 +226,16 @@ func writeOperation[T any](backgroundCtx context.Context, ctx context.Context, s
 	ready := make(chan result, 1)
 	ctx = context.WithValue(ctx, operationNameKey, operationName)
 	var res T
+	var defaultRes T
 	if err := s.cm.WriteOp(ctx, ready, func(c *client) error {
 		var opErr error
 		res, opErr = fn(c)
 		return opErr
 	}, cid.Cid{}); err != nil {
-		return res, err
+		return defaultRes, err
 	}
 	if err := waitResult(backgroundCtx, ctx, ready); err != nil {
-		return res, err
+		return defaultRes, err
 	}
 	return res, nil
 }

space/clientspace/keyvalueservice/service.go

@@ -61,6 +61,8 @@ type service struct {
 	lock          sync.RWMutex
 	subscriptions map[string]map[string]subscription
 
+	subscriptionBuf []subscription
+
 	keyValueStore keyvaluestorage.Storage
 	spaceCore     commonspace.Space
 	observer      keyvalueobserver.Observer
@@ -127,13 +129,20 @@ func (s *service) observeChanges(decryptor keyvaluestorage.Decryptor, kvs []inne
 			continue
 		}
 
+		// s.subscriptionBuf is safe to use without a lock because observeChanges runs only in one goroutine, and this buffer
+		// isn't used anywhere else
+		s.subscriptionBuf = s.subscriptionBuf[:0]
+
 		s.lock.RLock()
 		byKey := s.subscriptions[value.Key]
 		for _, sub := range byKey {
-			sub.observerFunc(value.Key, value)
+			s.subscriptionBuf = append(s.subscriptionBuf, sub)
 		}
 		s.lock.RUnlock()
 
+		for _, sub := range s.subscriptionBuf {
+			sub.observerFunc(value.Key, value)
+		}
 	}
 }