protondrive: don't cache a folder listing read while the folder changed

A listing whose pages were read before a file or folder was made in,
copied into or trashed from the folder cached what it read when it
finished, overwriting the change. As new files are added to a cached
listing rather than dropping it, the stale listing then lasted: a file
just uploaded stayed missing, and a trashed one stayed listed, until
something else dropped the listing.

The node cache now counts changes to the children of each folder being
listed, and a listing is cached only if none happened since its first
page was read. The count is kept only while a listing is in progress.

The fake server gains a Hold fault, which answers a request with the
response made on arrival only once released, to test this without
sleeps.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

backend/protondrive/cache.go

@@ -12,8 +12,17 @@ import (
 // nodes are shared and must not be modified.
 type nodeCache struct {
 	mu       sync.Mutex
-	nodes    map[string]*node    // by ID
-	listings map[string][]string // IDs of the complete listing of a folder, by folder ID
+	nodes    map[string]*node       // by ID
+	listings map[string][]string    // IDs of the complete listing of a folder, by folder ID
+	fetching map[string]*fetchState // folders being listed from the server, by folder ID
+}
+
+// fetchState tracks the listings of a folder being fetched, so that
+// one which began before a change to the folder's children isn't
+// cached over it.
+type fetchState struct {
+	count   int    // listings in progress
+	changes uint64 // changes to the folder's children since the first began
 }
 
 // newNodeCache returns an empty cache, or nil if enabled is false.
@@ -24,6 +33,7 @@ func newNodeCache(enabled bool) *nodeCache {
 	return &nodeCache{
 		nodes:    make(map[string]*node),
 		listings: make(map[string][]string),
+		fetching: make(map[string]*fetchState),
 	}
 }
 
@@ -46,8 +56,8 @@ func (c *nodeCache) put(n *node) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if old := c.nodes[n.ID]; old != nil && old.ParentID != n.ParentID {
-		delete(c.listings, old.ParentID)
-		delete(c.listings, n.ParentID)
+		c.childrenChangedLocked(old.ParentID)
+		c.childrenChangedLocked(n.ParentID)
 	}
 	c.nodes[n.ID] = n
 }
@@ -83,6 +93,11 @@ func (c *nodeCache) setChildren(parentID string, nodes []*node) {
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	c.setChildrenLocked(parentID, nodes)
+}
+
+// setChildrenLocked is setChildren with c.mu held.
+func (c *nodeCache) setChildrenLocked(parentID string, nodes []*node) {
 	ids := make([]string, len(nodes))
 	for i, n := range nodes {
 		ids[i] = n.ID
@@ -91,6 +106,53 @@ func (c *nodeCache) setChildren(parentID string, nodes []*node) {
 	c.listings[parentID] = ids
 }
 
+// startFetch records that a listing of folder parentID is being read
+// from the server and returns the token to pass to endFetch, which
+// must be called.
+func (c *nodeCache) startFetch(parentID string) (token uint64) {
+	if c == nil {
+		return 0
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	st := c.fetching[parentID]
+	if st == nil {
+		st = &fetchState{}
+		c.fetching[parentID] = st
+	}
+	st.count++
+	return st.changes
+}
+
+// endFetch ends a listing of folder parentID begun by startFetch with
+// token. If complete is set, nodes are cached as the listing, unless
+// the folder's children changed since it began, when what it read may
+// be out of date.
+func (c *nodeCache) endFetch(parentID string, token uint64, nodes []*node, complete bool) {
+	if c == nil {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	st := c.fetching[parentID]
+	if complete && st.changes == token {
+		c.setChildrenLocked(parentID, nodes)
+	}
+	st.count--
+	if st.count == 0 {
+		delete(c.fetching, parentID)
+	}
+}
+
+// childrenChangedLocked forgets the listing of folder parentID, and
+// marks its listings being fetched as out of date. c.mu must be held.
+func (c *nodeCache) childrenChangedLocked(parentID string) {
+	delete(c.listings, parentID)
+	if st := c.fetching[parentID]; st != nil {
+		st.changes++
+	}
+}
+
 // add caches n, a node new to its parent, and adds it to the cached
 // listing of its parent, if there is one.
 func (c *nodeCache) add(n *node) {
@@ -100,10 +162,13 @@ func (c *nodeCache) add(n *node) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if old := c.nodes[n.ID]; old != nil && old.ParentID != n.ParentID {
-		delete(c.listings, old.ParentID)
-		delete(c.listings, n.ParentID)
+		c.childrenChangedLocked(old.ParentID)
+		c.childrenChangedLocked(n.ParentID)
 	}
 	c.nodes[n.ID] = n
+	if st := c.fetching[n.ParentID]; st != nil {
+		st.changes++
+	}
 	if ids, ok := c.listings[n.ParentID]; ok && !slices.Contains(ids, n.ID) {
 		c.listings[n.ParentID] = append(ids, n.ID)
 	}
@@ -118,10 +183,16 @@ func (c *nodeCache) invalidate(id string) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if n := c.nodes[id]; n != nil {
-		delete(c.listings, n.ParentID)
+		c.childrenChangedLocked(n.ParentID)
+	} else {
+		// Its parent is unknown, and any listing being fetched may
+		// hold it.
+		for _, st := range c.fetching {
+			st.changes++
+		}
 	}
 	delete(c.nodes, id)
-	delete(c.listings, id)
+	c.childrenChangedLocked(id)
 }
 
 // invalidateChildren forgets the listing of folder parentID, but not
@@ -132,7 +203,7 @@ func (c *nodeCache) invalidateChildren(parentID string) {
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	delete(c.listings, parentID)
+	c.childrenChangedLocked(parentID)
 }
 
 // clear forgets everything.
@@ -144,4 +215,7 @@ func (c *nodeCache) clear() {
 	defer c.mu.Unlock()
 	clear(c.nodes)
 	clear(c.listings)
+	for _, st := range c.fetching {
+		st.changes++
+	}
 }

backend/protondrive/cache_internal_test.go

@@ -120,3 +120,65 @@ func TestNodeCacheAdd(t *testing.T) {
 
 	newNodeCache(false).add(a)
 }
+
+func TestNodeCacheFetch(t *testing.T) {
+	a := &node{ID: "a", ParentID: "dir"}
+	listed := func(c *nodeCache) bool {
+		_, ok := c.children("dir")
+		return ok
+	}
+
+	c := newNodeCache(true)
+	token := c.startFetch("dir")
+	c.endFetch("dir", token, []*node{a}, true)
+	assert.True(t, listed(c), "unchanged")
+	assert.Empty(t, c.fetching)
+
+	c = newNodeCache(true)
+	token = c.startFetch("dir")
+	c.endFetch("dir", token, []*node{a}, false)
+	assert.False(t, listed(c), "incomplete")
+	assert.Empty(t, c.fetching)
+
+	for name, change := range map[string]func(c *nodeCache){
+		"add":                func(c *nodeCache) { c.add(&node{ID: "b", ParentID: "dir"}) },
+		"invalidate child":   func(c *nodeCache) { c.invalidate("a") },
+		"invalidate unknown": func(c *nodeCache) { c.invalidate("unknown") },
+		"invalidate folder":  func(c *nodeCache) { c.invalidate("dir") },
+		"invalidateChildren": func(c *nodeCache) { c.invalidateChildren("dir") },
+		"put moved in":       func(c *nodeCache) { c.put(&node{ID: "other", ParentID: "dir"}) },
+		"put moved out":      func(c *nodeCache) { c.put(&node{ID: "a", ParentID: "elsewhere"}) },
+		"clear":              func(c *nodeCache) { c.clear() },
+	} {
+		t.Run(name, func(t *testing.T) {
+			c := newNodeCache(true)
+			c.put(a)
+			c.put(&node{ID: "other", ParentID: "elsewhere"})
+			first := c.startFetch("dir")
+			second := c.startFetch("dir")
+			change(c)
+			c.endFetch("dir", first, []*node{a}, true)
+			assert.False(t, listed(c), "first listing cached after a change")
+			// A fetch begun after the change is cached.
+			third := c.startFetch("dir")
+			c.endFetch("dir", second, []*node{a}, true)
+			assert.False(t, listed(c), "second listing cached after a change")
+			c.endFetch("dir", third, []*node{a}, true)
+			assert.True(t, listed(c))
+			assert.Empty(t, c.fetching, "fetch state kept")
+		})
+	}
+
+	// Changes elsewhere don't matter.
+	c = newNodeCache(true)
+	c.put(&node{ID: "x", ParentID: "other"})
+	token = c.startFetch("dir")
+	c.add(&node{ID: "y", ParentID: "other"})
+	c.invalidate("x")
+	c.invalidateChildren("other")
+	c.endFetch("dir", token, []*node{a}, true)
+	assert.True(t, listed(c))
+
+	var disabled *nodeCache
+	disabled.endFetch("dir", disabled.startFetch("dir"), nil, true)
+}

backend/protondrive/nodes.go

@@ -727,6 +727,9 @@ func (c *client) listChildren(ctx context.Context, parentID string, fn func([]*n
 	if err != nil {
 		return fmt.Errorf("list %s: %w", parentID, err)
 	}
+	// Taken before the first page, so a change made while the folder
+	// is read keeps what was read out of the cache.
+	token := c.cache.startFetch(parentID)
 
 	g, gCtx := errgroup.WithContext(ctx)
 	g.SetLimit(metadataConcurrency)
@@ -788,10 +791,10 @@ func (c *client) listChildren(ctx context.Context, parentID string, fn func([]*n
 	if waitErr := g.Wait(); waitErr != nil {
 		err = waitErr
 	}
+	c.cache.endFetch(parentID, token, all, err == nil)
 	if err != nil {
 		return fmt.Errorf("list %s: %w", parentID, err)
 	}
-	c.cache.setChildren(parentID, all)
 	return nil
 }
 

backend/protondrive/nodes_internal_test.go

@@ -1162,3 +1162,75 @@ func TestCreatedUnderUnverifiedFolderStaysUnverified(t *testing.T) {
 	require.NoError(t, err)
 	assert.Error(t, fresh.keys.unverified, "made read afresh")
 }
+
+// TestListingRacingChange checks a listing whose pages were read
+// before a change to the folder doesn't cache what it read: the next
+// listing shows the change, whether a file, folder or copy was made in
+// the folder or a child was trashed.
+func TestListingRacingChange(t *testing.T) {
+	t.Parallel()
+	for _, test := range []struct {
+		name    string
+		change  func(ctx context.Context, c *client, dirID, srcID, trashID string) error
+		present string // a name the folder must list after the change
+		absent  string // a name it must not
+	}{
+		{"upload", func(ctx context.Context, c *client, dirID, _, _ string) error {
+			_, err := c.upload(ctx, uploadRequest{ParentID: dirID, Name: "new", In: strings.NewReader("new"), Size: 3})
+			return err
+		}, "new", ""},
+		{"mkdir", func(ctx context.Context, c *client, dirID, _, _ string) error {
+			_, err := c.createFolder(ctx, dirID, "new", time.Time{})
+			return err
+		}, "new", ""},
+		{"copy", func(ctx context.Context, c *client, dirID, srcID, _ string) error {
+			src, err := c.getNode(ctx, srcID)
+			if err != nil {
+				return err
+			}
+			_, err = c.copyNode(ctx, src, dirID, "new")
+			return err
+		}, "new", ""},
+		{"trash", func(ctx context.Context, c *client, _, _, trashID string) error {
+			return c.trash(ctx, trashID)
+		}, "keep.txt", "trash.txt"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			c, srv, u := newTestClient(t)
+			ctx := context.Background()
+			dirID := u.SeedFolder(u.RootID(), "dir")
+			u.SeedFile(dirID, "keep.txt", []byte("keep"), time.Now())
+			trashID := u.SeedFile(dirID, "trash.txt", []byte("trash"), time.Now())
+			srcID := u.SeedFile(u.RootID(), "src.txt", []byte("src"), time.Now())
+			// Read before the listing, so its batch of children is the
+			// only links request held.
+			_, err := c.getNode(ctx, dirID)
+			require.NoError(t, err)
+			_, err = c.getNode(ctx, srcID)
+			require.NoError(t, err)
+
+			release := make(chan struct{})
+			held := make(chan struct{}, 1)
+			srv.Inject(protontest.Fault{Method: "POST", Pattern: "/drive/v2/volumes/{volumeID}/links", Times: 1, Hold: release, Held: held})
+			listed := make(chan error, 1)
+			go func() {
+				listed <- c.listChildren(ctx, dirID, func([]*node) error { return nil })
+			}()
+			<-held
+			require.NoError(t, test.change(ctx, c, dirID, srcID, trashID))
+			close(release)
+			require.NoError(t, <-listed)
+
+			got := listAll(t, c, dirID)
+			assert.Contains(t, got, test.present)
+			if test.absent != "" {
+				assert.NotContains(t, got, test.absent)
+			}
+			// The listing after the change is cached as usual.
+			srv.ResetRequests()
+			listAll(t, c, dirID)
+			assert.Zero(t, totalRequests(srv), "%v", srv.Requests())
+		})
+	}
+}

backend/protondrive/protontest/server.go

@@ -67,6 +67,13 @@ type Fault struct {
 	Delay        time.Duration // wait before handling the request
 	Gather       bool          // hold each faulted request until all Times have arrived
 	AfterHandler bool          // run the real handler, then replace its response
+
+	// Hold, if set, makes each faulted request run the real handler on
+	// arrival and get its response only once Hold is closed, so the
+	// response is as of arrival. Held, if set, is sent to once the
+	// response is ready and held.
+	Hold <-chan struct{}
+	Held chan<- struct{}
 }
 
 // fault is an injected Fault with the count of requests it has matched.
@@ -346,6 +353,24 @@ func (s *Server) serveFault(w http.ResponseWriter, r *http.Request, f *fault) {
 			return
 		}
 	}
+	if f.Hold != nil {
+		rec := httptest.NewRecorder()
+		s.mux.ServeHTTP(rec, r)
+		if f.Held != nil {
+			f.Held <- struct{}{}
+		}
+		select {
+		case <-f.Hold:
+		case <-r.Context().Done():
+			return
+		}
+		for k, v := range rec.Header() {
+			w.Header()[k] = v
+		}
+		w.WriteHeader(rec.Code)
+		_, _ = w.Write(rec.Body.Bytes())
+		return
+	}
 	if f.Status == 0 && !f.Drop {
 		s.mux.ServeHTTP(w, r)
 		return

backend/protondrive/protontest/server_test.go

@@ -224,6 +224,34 @@ func TestFaultInjection(t *testing.T) {
 		assert.Equal(t, http.StatusOK, resp.StatusCode)
 	})
 
+	t.Run("Hold", func(t *testing.T) {
+		s := New(t)
+		n := 0
+		s.handle("GET", "/tests/count", func(w http.ResponseWriter, r *http.Request) {
+			n++
+			writeJSON(w, http.StatusOK, map[string]int{"Code": api.CodeOK, "N": n})
+		})
+		release := make(chan struct{})
+		held := make(chan struct{}, 1)
+		s.Inject(Fault{Pattern: "/tests/count", Times: 1, Hold: release, Held: held})
+		first := make(chan []byte)
+		go func() {
+			_, body := do(t, s, "GET", "/tests/count", map[string]string{"x-pm-appversion": testAppVersion})
+			first <- body
+		}()
+		<-held
+		// The held request was handled on arrival; later ones pass.
+		_, body := do(t, s, "GET", "/tests/count", map[string]string{"x-pm-appversion": testAppVersion})
+		assert.JSONEq(t, `{"Code":1000,"N":2}`, string(body))
+		select {
+		case <-first:
+			t.Fatal("held request answered before release")
+		default:
+		}
+		close(release)
+		assert.JSONEq(t, `{"Code":1000,"N":1}`, string(<-first))
+	})
+
 	t.Run("AfterHandler", func(t *testing.T) {
 		s := New(t)
 		called := 0