.

wcfs/wcfs.go

@@ -607,7 +607,10 @@ type Watch struct {
 	// XXX locking
 
 	at     zodb.Tid	// requested to be watched @at
-	pinned SetI64   // blocks that are already pinned to be ≤ at
+//	pinned SetI64   // blocks that are already pinned to be ≤ at
+
+	// {} blk -> rev
+	pinned map[int64]zodb.Tid // blocks that are already pinned to be ≤ at
 }
 
 // -------- 3) Cache invariant --------
@@ -1318,13 +1321,14 @@ func (w *Watch) pin(ctx context.Context, blk int64, rev zodb.Tid) (err error) {
 
 	// XXX locking?
 	// XXX simultaneous calls?
+	// XXX rev=inf -> unpin to head
 
 	if rev > w.at {
 		panicf("f<%s>: watch%d: pin #%d @%s: watch.at (%s) < rev",
 			foid, w.link.id, blk, rev, w.at)
 	}
 
-	if w.pinned.Has(blk) {
+	if w.pinned[blk] == rev {
 		// XXX pinned has to be invalidated when w.at↑
 		return // already pinned
 	}
@@ -1338,7 +1342,7 @@ func (w *Watch) pin(ctx context.Context, blk int64, rev zodb.Tid) (err error) {
 		return fmt.Errorf("expect %q; got %q", "ack", ack)
 	}
 
-	w.pinned.Add(blk)
+	w.pinned[blk] = rev
 	return nil
 }
 
@@ -1382,7 +1386,8 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 		link:   wlink,
 		file:   f,
 		at:     at,
-		pinned: make(SetI64),
+		//pinned: make(SetI64),
+		pinned: make(map[int64]zodb.Tid),
 	}
 
 	// TODO register w to f here early, so that READs going in parallel to
@@ -1402,7 +1407,11 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 				continue
 			}
 
-			toPin[blk], _ = f.LastBlkRev(ctx, blk, at)
+			//toPin[blk], _ = f.LastBlkRev(ctx, blk, at)
+			pinrev, _ := f.LastBlkRev(ctx, blk, at)
+			if w.pinned[blk] != pinrev {	// XXX locking
+				toPin[blk] = pinrev
+			}
 		}
 	}