V0.13.x

Makefile

@@ -47,12 +47,15 @@ proto:
 	$(call generate_drpc,$(PKGMAP),commonspace/spacesyncproto/protos)
 	$(call generate_drpc,$(PKGMAP),commonspace/clientspaceproto/protos)
 	$(call generate_drpc,$(PKGMAP),commonfile/fileproto/protos)
+	$(call generate_drpc,,commonfile/fileproto/fileprotov2/protos)
+	$(call generate_drpc,,commonfile/fileproto/filep2p/protos)
 	$(call generate_drpc,$(PKGMAP),net/streampool/testservice/protos)
 	$(call generate_drpc,$(PKGMAP),net/endtoendtest/testpeer/testproto/protos)
 
 
 	$(call generate_drpc,,net/secureservice/handshake/handshakeproto/protos)
 	$(call generate_drpc,,net/rpc/limiter/limiterproto/protos)
+	$(call generate_drpc,,commonspace/pubsub/pubsubproto/protos)
 	$(call generate_drpc,$(PKGMAP),coordinator/coordinatorproto/protos)
 	$(call generate_drpc,,consensus/consensusproto/protos)
 	$(call generate_drpc,,identityrepo/identityrepoproto/protos)

app/ldiff/diff.go

@@ -1,7 +1,7 @@
 // Package ldiff provides a container of elements with fixed id and changeable content.
 // Diff can calculate the difference with another diff container (you can make it remote) with minimum hops and traffic.
 //
-//go:generate mockgen -destination mock_ldiff/mock_ldiff.go github.com/anyproto/any-sync/app/ldiff Diff,Remote,DiffContainer
+//go:generate mockgen -destination mock_ldiff/mock_ldiff.go github.com/anyproto/any-sync/app/ldiff Diff,Remote
 package ldiff
 
 import (

app/ldiff/hasher.go

@@ -0,0 +1,25 @@
+package ldiff
+
+import (
+	"encoding/hex"
+
+	"github.com/zeebo/blake3"
+)
+
+type Hasher struct {
+	hasher *blake3.Hasher
+}
+
+func (h *Hasher) HashId(id string) string {
+	h.hasher.Reset()
+	h.hasher.WriteString(id)
+	return hex.EncodeToString(h.hasher.Sum(nil))
+}
+
+func NewHasher() *Hasher {
+	return &Hasher{hashersPool.Get().(*blake3.Hasher)}
+}
+
+func ReleaseHasher(hasher *Hasher) {
+	hashersPool.Put(hasher.hasher)
+}

app/ocache/entry.go

@@ -23,10 +23,16 @@ type entry struct {
 	lastUsage time.Time
 	load      chan struct{}
 	loadErr   error
-	value     Object
-	close     chan struct{}
-	mx        sync.Mutex
-	cancel    context.CancelFunc
+	// loadAborted marks a failed load whose OWN context was already done
+	// when the loadFunc returned — the load was killed (its first caller
+	// went away, or the cache is closing), not refused by the loadFunc.
+	// Written by oCache.load before the load channel closes; read only
+	// after <-load, like loadErr.
+	loadAborted bool
+	value       Object
+	close       chan struct{}
+	mx          sync.Mutex
+	cancel      context.CancelFunc
 }
 
 func newEntry(id string, value Object, state entryState) *entry {
@@ -97,7 +103,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 +121,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 {