protondrive: make multi-thread downloads of files without block sizes linear

A file whose extended attributes don't record its block sizes, as some
other clients write, was read from the first block for every ranged
read, so a multi-thread copy, which opens each chunk as a range,
downloaded data quadratically: about 8.5 GiB for a 1 GiB file. The sizes
of blocks already read are now remembered per Fs (bounded to 16
revisions and 65536 blocks), so later chunks start at the right block.
A block list that disagrees with what was learnt is an integrity error.

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

backend/protondrive/download.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"slices"
 	"strconv"
 	"sync"
 
@@ -139,15 +140,20 @@ func (c *client) openRead(ctx context.Context, n *node, offset, limit int64, ful
 	if limit == 0 {
 		return io.NopCloser(bytes.NewReader(nil)), nil
 	}
-	// Without recorded block sizes nothing tells which block holds
-	// offset, as old clients didn't all write 4 MiB blocks, so the read
-	// starts at the first block and discards the bytes before offset.
+	// Without recorded block sizes only the sizes learnt by earlier
+	// reads tell which block holds offset, as old clients didn't all
+	// write 4 MiB blocks, so the read starts at the last block they
+	// reach and discards the bytes before offset.
 	first, skip, last := 1, offset, -1
-	if sizes != nil {
+	var startHash []byte
+	switch {
+	case sizes != nil:
 		first, skip = blockAt(sizes, offset)
 		if limit > 0 && !full {
 			last, _ = blockAt(sizes, offset+limit-1)
 		}
+	case !full:
+		first, skip, last, startHash = c.learnt.locate(n.RevisionID, offset, limit)
 	}
 	maxBlocks := len(sizes)
 	if sizes == nil {
@@ -162,6 +168,7 @@ func (c *client) openRead(ctx context.Context, n *node, offset, limit int64, ful
 		cancel:    cancel,
 		slots:     make(chan chan blockResult, downloadAhead),
 		sizes:     sizes,
+		startHash: startHash,
 		skip:      skip,
 		remaining: limit,
 	}
@@ -312,6 +319,7 @@ func (c *client) downloadBlock(ctx context.Context, bareURL, token string) (enc
 // blockResult is the outcome of a block of a download, or its end.
 type blockResult struct {
 	index int
+	hash  []byte   // the block's SHA-256
 	plain *pool.RW // the decrypted block; nil on error or at the end
 	err   error
 	end   bool // no more blocks follow; err is the verdict on the file
@@ -328,6 +336,9 @@ type blockReader struct {
 	slots  chan chan blockResult // results to read, in order
 	wg     sync.WaitGroup        // the lister and the fetches
 	sizes  []int64               // claimed plaintext block sizes; nil if not recorded
+	// startHash is the hash the first block had when its size was
+	// learnt, if it was.
+	startHash []byte
 
 	mu        sync.Mutex // held by Read and Close
 	cur       *pool.RW   // block being read
@@ -408,6 +419,11 @@ func (r *blockReader) listBlocks(first, last, maxBlocks int, verify bool) error
 			if err != nil {
 				return fmt.Errorf("block %d: %w", index, err)
 			}
+			if index == first && r.startHash != nil && !bytes.Equal(hash, r.startHash) {
+				// The sizes placing the range came from another block
+				// list of the same revision.
+				return fmt.Errorf("%w: block %d listed with another hash than when it was read before", errIntegrity, index)
+			}
 			if verify {
 				// Nothing is released before the manifest vouches for
 				// every block's hash and place.
@@ -498,7 +514,7 @@ func (r *blockReader) fetchBlock(b api.Block, hash []byte) blockResult {
 		var apiErr *api.Error
 		switch {
 		case err == nil:
-			return blockResult{index: index, plain: plain}
+			return blockResult{index: index, hash: hash, plain: plain}
 		case r.ctx.Err() != nil:
 			return blockResult{index: index, err: r.ctx.Err()}
 		case errors.As(err, &apiErr) && apiErr.Status == http.StatusNotFound && !refetched:
@@ -614,6 +630,12 @@ func (r *blockReader) next() error {
 		case res.end:
 			return io.EOF
 		}
+		if r.sizes == nil {
+			if err := r.c.learnt.learn(r.n.RevisionID, res.index, res.hash, res.plain.Size()); err != nil {
+				_ = res.plain.Close()
+				return err
+			}
+		}
 		if size := res.plain.Size(); r.skip > 0 && r.skip >= size {
 			r.skip -= size
 			_ = res.plain.Close()
@@ -661,3 +683,104 @@ func (r *blockReader) Close() error {
 
 // check interfaces
 var _ io.ReadCloser = (*blockReader)(nil)
+
+const (
+	// learntRevisions and learntBlocks bound the block sizes a client
+	// learns: of at most this many revisions, and this many blocks in
+	// all, about 2.5 MiB.
+	learntRevisions = 16
+	learntBlocks    = 1 << 16
+)
+
+// learntBlock is the plaintext size of a block read, with its hash.
+type learntBlock struct {
+	hash [sha256.Size]byte
+	size int64
+}
+
+// blockSizeCache holds the plaintext sizes of the blocks, from the
+// first on, read of revisions which don't record their block sizes, so
+// a later ranged read, as each chunk of a multi-thread copy is, can
+// start at the block holding its offset. The zero value is empty.
+type blockSizeCache struct {
+	mu     sync.Mutex
+	revs   map[string][]learntBlock // by revision ID
+	order  []string                 // revision IDs, oldest first
+	blocks int                      // learnt in all
+}
+
+// locate returns where a read of limit bytes (to the end if negative)
+// from offset of revision id starts, as for blockReader: the first
+// block, the bytes of it to skip, the last block or -1 if not known,
+// and the hash the first block had when its size was learnt, or nil if
+// it wasn't.
+func (b *blockSizeCache) locate(id string, offset, limit int64) (first int, skip int64, last int, startHash []byte) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	blocks := b.revs[id]
+	first, skip, last = len(blocks)+1, offset, -1
+	for i, lb := range blocks {
+		if skip < lb.size {
+			first = i + 1
+			startHash = slices.Clone(lb.hash[:])
+			break
+		}
+		skip -= lb.size
+	}
+	if first > len(blocks) || limit <= 0 {
+		return first, skip, last, startHash
+	}
+	end := skip + limit - 1
+	for i := first - 1; i < len(blocks); i++ {
+		if end < blocks[i].size {
+			return first, skip, i + 1, startHash
+		}
+		end -= blocks[i].size
+	}
+	return first, skip, last, startHash
+}
+
+// learn records that block index of revision id, whose SHA-256 is hash,
+// is size bytes. A block learnt before with another hash or size is
+// errIntegrity: the server listed two blocks for one place.
+func (b *blockSizeCache) learn(id string, index int, hash []byte, size int64) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	blocks, ok := b.revs[id]
+	if index <= len(blocks) {
+		if lb := blocks[index-1]; !bytes.Equal(lb.hash[:], hash) || lb.size != size {
+			return fmt.Errorf("%w: block %d listed with another hash than when it was read before", errIntegrity, index)
+		}
+		return nil
+	}
+	if index != len(blocks)+1 || len(hash) != sha256.Size {
+		return nil
+	}
+	if !ok {
+		if b.revs == nil {
+			b.revs = make(map[string][]learntBlock)
+		}
+		for len(b.order) >= learntRevisions {
+			b.forgetOldest()
+		}
+		b.order = append(b.order, id)
+	}
+	for b.blocks >= learntBlocks && len(b.order) > 1 && b.order[0] != id {
+		b.forgetOldest()
+	}
+	if b.blocks >= learntBlocks {
+		return nil
+	}
+	lb := learntBlock{size: size}
+	copy(lb.hash[:], hash)
+	b.revs[id] = append(blocks, lb)
+	b.blocks++
+	return nil
+}
+
+// forgetOldest drops the revision learnt first.
+func (b *blockSizeCache) forgetOldest() {
+	b.blocks -= len(b.revs[b.order[0]])
+	delete(b.revs, b.order[0])
+	b.order = b.order[1:]
+}

backend/protondrive/download_internal_test.go

@@ -139,29 +139,27 @@ func TestDownloadRanges(t *testing.T) {
 	claimed := seedFile(t, c, u, "big", content)
 	unrecorded := withoutBlockSizes(claimed)
 	for _, tc := range []struct {
-		offset, limit    int64
-		blocks           int // blocks fetched with the block sizes claimed
-		blocksUnrecorded int // fewest blocks fetched without them, read from the start
+		offset, limit int64
+		blocks        int // blocks fetched with the block sizes claimed
 	}{
-		{0, -1, 3, 3},
-		{0, 1, 1, 1},
-		{0, 0, 0, 0},
-		{4<<20 - 1, 2, 2, 2},
-		{4 << 20, -1, 2, 3},
-		{bigSize - 1, -1, 1, 3},
-		{bigSize, -1, 0, 3},
-		{8 << 20, 100, 1, 3},
-		{bigSize - 10, 100, 1, 3},
+		{0, -1, 3},
+		{0, 1, 1},
+		{0, 0, 0},
+		{4<<20 - 1, 2, 2},
+		{4 << 20, -1, 2},
+		{bigSize - 1, -1, 1},
+		{bigSize, -1, 0},
+		{8 << 20, 100, 1},
+		{bigSize - 10, 100, 1},
 	} {
 		want := content[tc.offset:]
 		if tc.limit >= 0 {
 			want = want[:min(tc.limit, int64(len(want)))]
 		}
 		for _, v := range []struct {
-			name   string
-			n      *node
-			blocks int
-		}{{"claimed", claimed, tc.blocks}, {"unrecorded", unrecorded, tc.blocksUnrecorded}} {
+			name string
+			n    *node
+		}{{"claimed", claimed}, {"unrecorded", unrecorded}} {
 			t.Run(v.name+"/"+strconv.FormatInt(tc.offset, 10)+"+"+strconv.FormatInt(tc.limit, 10), func(t *testing.T) {
 				srv.ResetRequests()
 				data, err := readRange(t, c, v.n, tc.offset, tc.limit)
@@ -169,13 +167,14 @@ func TestDownloadRanges(t *testing.T) {
 				assert.True(t, bytes.Equal(want, data), "got %d bytes, want %d", len(data), len(want))
 				fetched := srv.Requests()[storagePattern]
 				if v.n == claimed {
-					assert.Equal(t, v.blocks, fetched, "blocks fetched")
+					assert.Equal(t, tc.blocks, fetched, "blocks fetched")
 					return
 				}
-				// Blocks past the range may be fetched ahead before
-				// the reader is closed.
-				assert.GreaterOrEqual(t, fetched, v.blocks, "blocks fetched")
-				assert.LessOrEqual(t, fetched, min(v.blocks, 1)*3, "blocks fetched")
+				// The block sizes learnt by the first, full, read
+				// place the range; blocks past it may be fetched ahead
+				// before the reader is closed.
+				assert.GreaterOrEqual(t, fetched, tc.blocks, "blocks fetched")
+				assert.LessOrEqual(t, fetched, 3, "blocks fetched")
 			})
 		}
 	}
@@ -213,6 +212,47 @@ func TestDownloadRangesUnrecordedBlockSizes(t *testing.T) {
 	}
 }
 
+func TestDownloadChunksUnrecordedBlockSizes(t *testing.T) {
+	t.Parallel()
+	c, srv, u := newTestClient(t)
+	const (
+		blocks     = 32
+		chunk      = 4 << 20
+		blockBytes = 1 << 20
+	)
+	content := testContent(blocks * blockBytes)
+	id := u.SeedFileInBlocks(u.RootID(), "chunks", content, blockBytes)
+	n := getTestNode(t, c, id)
+	require.Nil(t, n.X.BlockSizes)
+
+	// A multi-thread copy opens each chunk as a range of its own.
+	srv.ResetRequests()
+	for offset := int64(0); offset < int64(len(content)); offset += chunk {
+		data, err := readRange(t, c, n, offset, chunk)
+		require.NoError(t, err)
+		require.True(t, bytes.Equal(content[offset:offset+chunk], data), "chunk at %d", offset)
+	}
+	fetched := srv.Requests()[storagePattern]
+	t.Logf("%d blocks fetched for %d chunks of a file of %d blocks", fetched, len(content)/chunk, blocks)
+	// Reading each chunk from the first block would fetch about
+	// blocks^2/(2*blocks per chunk) = 128; each chunk may fetch up to
+	// downloadAhead blocks past its end.
+	assert.LessOrEqual(t, fetched, blocks+len(content)/chunk*downloadAhead, "chunks read from the first block")
+
+	// Once the sizes are known, a range fetches only its own blocks.
+	srv.ResetRequests()
+	data, err := readRange(t, c, n, 10*blockBytes+5, 2*blockBytes)
+	require.NoError(t, err)
+	assert.Equal(t, content[10*blockBytes+5:12*blockBytes+5], data)
+	assert.Equal(t, 3, srv.Requests()[storagePattern], "blocks fetched")
+
+	// A block list which changed since then is an integrity error.
+	u.Corrupt(id, "swap-blocks")
+	_, err = readRange(t, c, n, 5, 10)
+	require.ErrorIs(t, err, errIntegrity)
+	assert.Contains(t, err.Error(), "listed with another hash")
+}
+
 func TestDownloadLegacyFile(t *testing.T) {
 	t.Parallel()
 	c, _, u := newTestClient(t)

backend/protondrive/nodes.go

@@ -101,6 +101,7 @@ type client struct {
 	uploadTokens   *pacer.TokenDispenser
 	downloadTokens *pacer.TokenDispenser
 	cache          *nodeCache
+	learnt         blockSizeCache // block sizes of files which don't record them
 
 	// verifierHook, if set, may change each block's verifier token
 	// before it is sent; for tests.

docs/TODO.md

@@ -130,9 +130,15 @@ Small, specific, independently actionable. Remove only when actually done.
   so such fixtures can be generated from a worktree of it.
 
 - [ ] protondrive: a ranged read of a file which doesn't record its
-  block sizes (written by some other clients) downloads from the first
-  block every time, so random-offset reads of large such files through a
-  mount are slow. Block sizes learnt on a full read could be cached.
+  block sizes (written by some other clients) starts at the furthest
+  block whose size an earlier read in the same `Fs` learnt
+  (download.go `blockSizeCache`, 16 revisions, 65536 blocks), else at
+  the first block. The first chunks of a multi-thread copy, which start
+  together, still each read from the first block, and a random-offset
+  read beyond what was read before through a mount is slow. A ranged
+  read with sizes unknown also fetches up to `downloadAhead` blocks past
+  its end. Neither cache nor listing tells where a block starts without
+  reading the ones before it, as the packet framing varies by writer.
 
 - [ ] protondrive: `client.apiCall` retries POST and PUT mutations
   (create folder, move, rename, copy) after 5xx and network errors, as the

docs/content/protondrive.md

@@ -205,10 +205,12 @@ manifest, so a malicious server could serve validly encrypted blocks of
 the same file in the wrong place, or from an older version of the same
 size. Proton's SDK has the same limit for ranged reads. A range which
 covers the whole file is checked as a full read. A file uploaded by a
-client which doesn't record its block sizes can't be entered in the
-middle, so a ranged read of it downloads, checks and discards every
-block before the range, which makes it slower; its manifest still isn't
-checked.
+client which doesn't record its block sizes can only be entered in the
+middle as far as earlier reads in the same rclone process have reached:
+beyond that a ranged read downloads, checks and discards every block
+before the range, which makes it slower. rclone remembers the sizes of
+up to 65536 such blocks, so each chunk of a multi-thread download starts
+near where an earlier one ended. Its manifest still isn't checked.
 
 **Trust is rooted in your own addresses.** The share holding your files
 must have been created, and its root folder signed, by one of your