protondrive: fix uploads failing while Proton catches up after a commit

Proton's server can take a moment to show a change: a file read back
just after its commit may be missing or still a draft with no active
revision. Uploads then failed with "upload committed, but the file now
has another revision (revision  is active ...)" or "object not found".

The read back after a commit, the read of a server-side copy and, with
enable_caching off, the read of a folder to write into now retry with
backoff for about 7 s while the change doesn't show. A committed upload
the server still doesn't show after that fails with a new "committed,
but the server doesn't show it yet" error, deletes nothing and is not
retried at once. The fake server gains SetReadLag to model the lag.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Lucas WintherandClaude Opus 5.5 committed 2026-11-12 11:57:02 +01:00
1 parent a2dc6f71f1
commit af6fdd7899
12 files changed
+288 -13

No files matched your search

+45 -1
View File
@@ -56,6 +56,14 @@ const (
// maxRetryAfter is the longest Retry-After honoured.
maxRetryAfter = 60 * time.Second
// readBackAttempts is how many times a node just changed is read
// while the server doesn't show the change yet. With the first wait
// readBackDelay, doubled each time up to readBackMaxDelay, they span
// about 7 s.
readBackAttempts = 8
readBackDelay = 100 * time.Millisecond
readBackMaxDelay = 2 * time.Second
// nameSignedSince and hashKeySignedSince are the Unix times from
// which names and hash keys are signed: as in the SDK, a missing
// signature on an older one is not an error.
@@ -100,6 +108,8 @@ type client struct {
maxFileBlocks int // the most blocks a file may list: maxFileBlocks, unless a test lowers it
maxFolderDepth int // the most ancestors fetched for a node: maxFolderDepth, unless a test lowers it
readBackDelay time.Duration // first wait before reading a changed node again: readBackDelay, unless a test lowers it
mu sync.Mutex
folderKeys map[string]*nodeKeys // by folder ID; kept whatever the cache, as node keys never change
reported map[string]bool // problems already logged, by kind and node ID
@@ -191,6 +201,7 @@ func newClient(ctx context.Context, s *session, opt clientOptions) (*client, err
reported: make(map[string]bool),
maxFileBlocks: maxFileBlocks,
maxFolderDepth: maxFolderDepth,
readBackDelay: readBackDelay,
}
if err := c.bootstrap(ctx); err != nil {
c.close()
@@ -714,6 +725,33 @@ func (c *client) fetchNodeIn(ctx context.Context, id string, below []string) (*n
return n, nil
}
// readBack reads node id, just changed, from the server as fetchNode
// does, reading it again while shows reports the read doesn't show the
// change yet: Proton's servers can take a moment to. It returns the
// last read once shows reports true, after readBackAttempts reads, or
// when ctx is done.
func (c *client) readBack(ctx context.Context, id string, shows func(n *node, err error) bool) (*node, error) {
delay := c.readBackDelay
for attempt := 1; ; attempt++ {
n, err := c.fetchNode(ctx, id)
if shows(n, err) || attempt >= readBackAttempts {
return n, err
}
fs.Debugf(nil, "protondrive: node %s doesn't show its change yet: reading it again in %v", id, delay)
select {
case <-time.After(delay):
case <-ctx.Done():
return n, err
}
delay = min(2*delay, readBackMaxDelay)
}
}
// found reports whether a read of a node which returned err found it.
func found(_ *node, err error) bool {
return !errors.Is(err, errNotFound)
}
// activeNode reads node id from the server whatever the cache holds,
// and returns it if it is active. Otherwise it forgets the node and
// the listing of its parent and returns errNotFound.
@@ -926,6 +964,12 @@ func mutationError(what string, err error) error {
// IgnoreSignatureErrors is set.
func (c *client) targetFolder(ctx context.Context, id string) (*node, error) {
n, err := c.getNode(ctx, id)
if errors.Is(err, errNotFound) && c.cache == nil {
// Without the cache a folder just made is read from the
// server, which may not show it yet. With it, the folder was
// cached when made, so a miss means it is gone.
n, err = c.readBack(ctx, id, found)
}
switch {
case err != nil:
return nil, err
@@ -1267,7 +1311,7 @@ func (c *client) copyNode(ctx context.Context, n *node, newParentID, newName str
c.cache.invalidateChildren(newParentID)
return nil, mutationError(what, err)
}
cp, err := c.getNode(ctx, resp.LinkID)
cp, err := c.readBack(ctx, resp.LinkID, found)
if err != nil {
c.cache.invalidateChildren(newParentID)
return nil, fmt.Errorf("%s: %w", what, err)
@@ -1283,3 +1283,19 @@ func TestFolderDepthLimit(t *testing.T) {
_, err = c.getNode(context.Background(), ids[5])
require.NoError(t, err)
}
// TestCopyReadBackLag copies a file while the server takes a moment to
// show the copy.
func TestCopyReadBackLag(t *testing.T) {
t.Parallel()
c, srv, u := newTestClient(t)
ctx := context.Background()
fileID := u.SeedFile(u.RootID(), "f.txt", []byte("content"), time.Now())
file, err := c.getNode(ctx, fileID)
require.NoError(t, err)
srv.SetReadLag(2, true)
cp, err := c.copyNode(ctx, file, u.RootID(), "copy.txt")
require.NoError(t, err)
assert.Equal(t, "copy.txt", cp.Name)
assert.Equal(t, file.sha1(), cp.sha1())
}
+1 -1
View File
@@ -363,7 +363,7 @@ func translateError(err error, op errorOp) error {
// The upload landed; doing it again would only replace the
// other client's newer revision.
return fserrors.NoRetryError(err)
case errors.Is(err, errUploadOutcomeUnknown):
case errors.Is(err, errUploadOutcomeUnknown), errors.Is(err, errUploadUnconfirmed):
// An immediate retry could upload a revision which already
// committed. A later pass of a sync reads the file first, so
// may still retry.
@@ -89,12 +89,14 @@ func TestTranslateError(t *testing.T) {
assert.ErrorIs(t, got, err)
})
}
t.Run("outcome unknown no low level retry", func(t *testing.T) {
got := translateError(fmt.Errorf("op: %w", errUploadOutcomeUnknown), opObject)
assert.True(t, fserrors.IsNoLowLevelRetryError(got))
assert.False(t, fserrors.IsNoRetryError(got))
assert.ErrorIs(t, got, errUploadOutcomeUnknown)
})
for _, err := range []error{errUploadOutcomeUnknown, errUploadUnconfirmed} {
t.Run(err.Error()+" no low level retry", func(t *testing.T) {
got := translateError(fmt.Errorf("op: %w", err), opObject)
assert.True(t, fserrors.IsNoLowLevelRetryError(got))
assert.False(t, fserrors.IsNoRetryError(got))
assert.ErrorIs(t, got, err)
})
}
}
// sessionConfig returns the config keys of a v1.76 remote holding a new
@@ -624,3 +626,14 @@ func TestOpenRange(t *testing.T) {
})
}
}
// TestPutNestedReadBackLag puts a file into new nested directories
// while the server takes a moment to show new nodes.
func TestPutNestedReadBackLag(t *testing.T) {
srv := protontest.New(t)
u := srv.AddUser("user", "pass")
f := newTestFs(t, sessionConfig(srv, u))
srv.SetReadLag(2, true)
o := putFile(t, f, "A1/B1/C1/one", []byte("aaa"))
assert.Equal(t, int64(3), o.Size())
}
+44 -2
View File
@@ -1702,13 +1702,55 @@ func (s *Server) handleLinks(w http.ResponseWriter, r *http.Request, u *User, d
}
resp := &linksResponse{Links: []linkDetailsJSON{}, Code: api.CodeOK}
for _, id := range req.LinkIDs {
if l := d.links[id]; l != nil {
resp.Links = append(resp.Links, d.linkDetails(l))
l := d.links[id]
if l == nil {
continue
}
details := d.linkDetails(l)
if v := s.lagging[id]; v != nil {
if v.reads--; v.reads <= 0 {
delete(s.lagging, id)
}
if v.before == nil {
continue
}
details = *v.before
}
resp.Links = append(resp.Links, details)
}
writeJSON(w, http.StatusOK, resp)
}
// lagView is what reads of a node see until its change shows.
type lagView struct {
reads int // reads left which don't see the change
before *linkDetailsJSON // the node before the change; nil leaves it out
}
// SetReadLag makes the first reads reads of a node through the links
// route, after it is created, copied or has a revision committed, not
// see the change, as Proton's servers may not for a moment: they leave
// the node out if it is new or missing is set, and otherwise show it as
// it was before. Zero, the default, turns the lag off.
func (s *Server) SetReadLag(reads int, missing bool) {
s.mu.Lock()
defer s.mu.Unlock()
s.readLag, s.readLagMissing = reads, missing
}
// lagLocked starts the read lag, if any, of node id after a change;
// before is its wire form before the change, or nil if it is new.
// srv.mu must be held.
func (s *Server) lagLocked(id string, before *linkDetailsJSON) {
if s.readLag <= 0 {
return
}
if s.readLagMissing {
before = nil
}
s.lagging[id] = &lagView{reads: s.readLag, before: before}
}
// handleChildren returns a page of the IDs of the active children of a
// folder in creation order, after AnchorID if given.
func (s *Server) handleChildren(w http.ResponseWriter, r *http.Request, u *User, d *drive) {
+2
View File
@@ -497,6 +497,7 @@ func (s *Server) handleCreateFolder(w http.ResponseWriter, r *http.Request, u *U
l.xattr = *req.XAttr
}
d.addLink(l)
s.lagLocked(l.id, nil)
writeJSON(w, http.StatusOK, &createFolderResponse{Folder: folderIDJSON{ID: l.id}, Code: api.CodeOK})
}
@@ -647,6 +648,7 @@ func (s *Server) handleCopy(w http.ResponseWriter, r *http.Request, u *User, d *
c.name, c.nameHash, c.nameSignatureEmail = req.Name, req.Hash, *req.NameSignatureEmail
c.nodePass = sealedPassphrase{armored: req.NodePassphrase, signature: sig}
c.signatureEmail = signer
s.lagLocked(c.id, nil)
writeJSON(w, http.StatusOK, &copyResponse{LinkID: c.id, Code: api.CodeOK})
}
+4
View File
@@ -115,6 +115,9 @@ type Server struct {
storagePeak int // most block uploads in progress at once
secrets []issuedSecret // every secret issued, in order
keysAs map[string]*User // whose keys GET /core/v4/keys/all serves for an email, overriding the owner's
readLag int // reads of a node after a change which don't see it; see SetReadLag
readLagMissing bool // lagging reads of a changed node leave it out
lagging map[string]*lagView // nodes whose reads lag, by ID
}
// New starts a fake server which is closed when the test ends.
@@ -155,6 +158,7 @@ func newServer() *Server {
downloads: make(map[string]*download),
unlockedKeys: make(map[string]*crypto.Key),
uploads: make(map[string]*uploadURL),
lagging: make(map[string]*lagView),
}
s.handle("GET", "/tests/ping", func(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]int{"Code": api.CodeOK})
+2
View File
@@ -582,6 +582,7 @@ func (s *Server) handleCommit(w http.ResponseWriter, r *http.Request, u *User, d
writeInvalid(w, err)
return
}
before := d.linkDetails(l)
for _, other := range l.revisions {
if other.state == revisionStateActive {
other.state = revisionStateObsolete
@@ -593,6 +594,7 @@ func (s *Server) handleCommit(w http.ResponseWriter, r *http.Request, u *User, d
rev.blocks, rev.pending, rev.verificationCode = blocks, nil, nil
l.state, l.modifyTime = linkStateActive, now
s.dropUploadsLocked(rev)
s.lagLocked(l.id, &before)
writeJSON(w, http.StatusOK, &codeJSON{Code: api.CodeOK})
}
+30 -3
View File
@@ -38,6 +38,10 @@ var errUploadOutcomeUnknown = errors.New("upload outcome unknown: it may have su
// file read back has another revision, which is not this upload's.
var errUploadSuperseded = errors.New("upload committed, but the file now has another revision")
// errUploadUnconfirmed is returned when an upload was committed but the
// server still didn't show it when the file was last read back.
var errUploadUnconfirmed = errors.New("upload committed, but the server doesn't show it yet")
const (
// defaultUploadConcurrency and maxUploadConcurrency bound the blocks
// of one file being uploaded at once, as in the SDK.
@@ -99,8 +103,10 @@ type draft struct {
// or the revision the upload started from still active). Otherwise
// nothing is deleted and a failed commit is errUploadOutcomeUnknown: the
// upload may have succeeded. A commit which succeeded but whose file has
// another revision by the time it is read back is errUploadSuperseded.
// The returned node is always the upload's own revision.
// another revision by the time it is read back is errUploadSuperseded,
// and one the server still doesn't show after a few reads is
// errUploadUnconfirmed. The returned node is always the upload's own
// revision.
func (c *client) upload(ctx context.Context, r uploadRequest) (*node, error) {
what := fmt.Sprintf("upload %q", r.Name)
if r.Existing != nil {
@@ -392,7 +398,9 @@ func (c *client) uploadDraft(ctx context.Context, d *draft, r uploadRequest, sig
// of a commit which took effect would delete the file or revision.
readCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), draftCleanupTimeout)
defer cancel()
n, err = c.fetchNode(readCtx, d.nodeID)
n, err = c.readBack(readCtx, d.nodeID, func(n *node, err error) bool {
return !d.lagging(n, err, commitErr == nil)
})
ours := err == nil && n.State == api.LinkStateActive && n.RevisionID == d.revisionID
if ours {
c.cache.add(n)
@@ -407,6 +415,9 @@ func (c *client) uploadDraft(ctx context.Context, d *draft, r uploadRequest, sig
return n, true, nil
case ours:
return n, true, nil
case commitErr == nil && d.lagging(n, err, true):
fs.Debugf(nil, "protondrive: node %s: revision %s committed, but the server doesn't show it yet", d.nodeID, d.revisionID)
return nil, true, errUploadUnconfirmed
case commitErr == nil && err != nil:
return nil, true, fmt.Errorf("read the uploaded file: %w", err)
case commitErr == nil:
@@ -422,6 +433,22 @@ func (c *client) uploadDraft(ctx context.Context, d *draft, r uploadRequest, sig
return nil, true, fmt.Errorf("commit: %w: %w (the file has changed since)", errUploadOutcomeUnknown, commitErr)
}
// lagging reports whether n and err, read after the commit of d, may
// be from before the server shows the commit: the node isn't found, or
// is active without an active revision, or, if the commit is known to
// have succeeded, looks as it did before it.
func (d *draft) lagging(n *node, err error, committed bool) bool {
switch {
case errors.Is(err, errNotFound):
return true
case err != nil:
return false
case n.State == api.LinkStateActive && n.RevisionID == "":
return true
}
return committed && d.notCommitted(n)
}
// notCommitted reports whether n, read after a failed commit of d,
// proves the commit didn't take effect: a new file still a draft, or a
// file whose active revision is still the one d started from.
@@ -6,6 +6,7 @@ import (
"crypto/sha1"
"encoding/hex"
"errors"
"fmt"
"io"
"net/http"
"strconv"
@@ -753,3 +754,86 @@ func TestUploadRevisionCommitRefused(t *testing.T) {
assert.Equal(t, 1, srv.Requests()[deleteRevisionRequest])
requireUploaded(t, u, map[string][]byte{"f": []byte("first")})
}
// TestUploadReadBackLag uploads while the server takes a moment to show
// what was committed, as Proton's does: the upload still succeeds with
// its own revision and nothing is deleted.
func TestUploadReadBackLag(t *testing.T) {
t.Parallel()
content := []byte("content")
for _, tc := range []struct {
name string
missing bool
existing bool
}{
{name: "NewFileStale"},
{name: "NewFileMissing", missing: true},
{name: "NewRevisionStale", existing: true},
{name: "NewRevisionMissing", existing: true, missing: true},
} {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
c, srv, u := newUploadClient(t, clientOptions{})
r := uploadRequest{ParentID: c.rootID(), Name: "f", In: bytes.NewReader(content), Size: int64(len(content)), ModTime: testModTime}
var first *node
if tc.existing {
first = uploadBytes(t, c, "f", []byte("first"))
r.Existing = first
}
srv.ResetRequests()
srv.SetReadLag(2, tc.missing)
n, err := c.upload(context.Background(), r)
require.NoError(t, err)
assert.NotEmpty(t, n.RevisionID)
if first != nil {
assert.NotEqual(t, first.RevisionID, n.RevisionID)
}
assert.Equal(t, sha1Of(content), n.sha1())
requireNoDeletes(t, srv)
requireUploaded(t, u, map[string][]byte{"f": content})
})
}
}
// TestUploadIntoNewFolderReadBackLag creates nested folders and uploads
// into the deepest without a cache while the server takes a moment to
// show new nodes.
func TestUploadIntoNewFolderReadBackLag(t *testing.T) {
t.Parallel()
c, srv, u := newTestClientOpt(t, clientOptions{DraftClientUID: testDraftClientUID})
srv.SetReadLag(2, true)
ctx := context.Background()
parentID := c.rootID()
for _, name := range []string{"A1", "B1", "C1"} {
n, err := c.createFolder(ctx, parentID, name, time.Time{})
require.NoError(t, err)
parentID = n.ID
}
content := []byte("aaa")
_, err := c.upload(ctx, uploadRequest{ParentID: parentID, Name: "one", In: bytes.NewReader(content), Size: int64(len(content)), ModTime: testModTime})
require.NoError(t, err)
requireNoDeletes(t, srv)
requireUploaded(t, u, map[string][]byte{"A1/B1/C1/one": content})
}
// TestUploadReadBackLagOutlasted commits an upload which the server
// doesn't show for longer than the client reads it back: the upload
// fails as unconfirmed and nothing is deleted.
func TestUploadReadBackLagOutlasted(t *testing.T) {
t.Parallel()
for _, missing := range []bool{false, true} {
t.Run(fmt.Sprintf("missing=%v", missing), func(t *testing.T) {
t.Parallel()
c, srv, u := newUploadClient(t, clientOptions{})
c.readBackDelay = time.Millisecond
srv.SetReadLag(readBackAttempts, missing)
n, err := uploadContent(context.Background(), c, "f", strings.NewReader("mine"), 4)
assert.ErrorIs(t, err, errUploadUnconfirmed)
assert.NotErrorIs(t, err, errUploadSuperseded)
assert.Nil(t, n)
assert.Equal(t, readBackAttempts, srv.Requests()["POST /drive/v2/volumes/{volumeID}/links"])
requireNoDeletes(t, srv)
requireUploaded(t, u, map[string][]byte{"f": []byte("mine")})
})
}
}
+27
View File
@@ -5,6 +5,33 @@ it with a new one that links back. One entry per decision a future maintainer
would otherwise have to reverse-engineer.
No index: `grep -n '^## ' docs/DECISIONS.md` lists every entry.
## 2026-11-12 - Proton Drive uploads tolerate read-after-write lag, but never invent the result
**Status:** accepted
**Context:** Against the real service an upload whose commit succeeded
failed because the node read back just after was missing or still a
draft with an empty active revision (`errUploadSuperseded` with revision
"", and "object not found"). The client read the node once and trusted it.
**Decision:** After a commit the node is read again with backoff (100 ms
doubling to 2 s, 8 reads, about 7 s) while it is missing, active with no
revision, or - when the commit returned success - still as it was before
the commit. `errUploadSuperseded` is kept for a non-empty revision which
is neither ours nor the one we started from. If the lag outlasts the
reads, the upload returns `errUploadUnconfirmed` (no low-level retry, as
for `errUploadOutcomeUnknown`), deletes nothing, and a later sync pass
finds the file. The same not-found retry covers a copy's read back and,
with `enable_caching` off, a folder to write into (with it on, a folder
the client made is cached, so a miss means it is gone).
**Alternatives:** Returning success with a node built from what was
committed (the caller then reads, moves or updates through it, and that
needs the decrypted keys and server metadata the client would be
inventing; a wrong node would be cached and acted on); retrying every
read after any write (slows every genuine not-found).
**Consequences:** With `enable_caching` off, a put into a folder deleted
elsewhere takes about 7 s longer to fail. A lag longer than 7 s still fails the upload,
now with an error saying it was committed. See `docs/TODO.md` for the
read-after-write spots not yet tolerant.
## 2026-11-12 - Proton Drive golden fixtures are re-keyed captures
**Status:** accepted
+14
View File
@@ -151,6 +151,20 @@ Small, specific, independently actionable. Remove only when actually done.
- [ ] protondrive: confirm the HTTP status and code Proton returns for a
bad verifier token, a missing block and a stale `CurrentRevisionID`
(the fake uses 422 / 2000, which the spec maps to "upgrade rclone").
- [ ] protondrive: Proton's server can take a moment to show a change
(seen live: a node read just after its commit was missing, or still a
draft with no active revision). The fake models this only through
`Server.SetReadLag`, on the links route after a create, copy or
commit; how long the real lag lasts and whether listings
(`folders/{id}/children`) lag too is unmeasured. Only the reads back
after a commit, a copy and, with `enable_caching` off, of a folder to
write into (`targetFolder`) retry, for about 7 s. Not tolerant yet, and the next
symptoms to expect: `activeNode` (Update, Move, Copy, DirMove of an
object just written) fails "object not found" while the node is
missing or still a draft; Move's read of the moved node may be stale;
after a commit that *failed*, a stale read showing the draft or the old
revision still counts as proof the commit didn't land and the draft is
deleted (retrying there would slow every failed commit by 7 s).
- [ ] protondrive: several `Fs` of one account in one process each have
their own node cache; only Move/DirMove sources are invalidated across