diff --git a/cache.go b/cache.go index cd4fe24..8898dd0 100644 --- a/cache.go +++ b/cache.go @@ -208,7 +208,7 @@ func (protonDrive *ProtonDrive) getLink(ctx context.Context, linkID string) (*pr } // no cached data, fetch - link, err := protonDrive.c.GetLink(ctx, protonDrive.MainShare.ShareID, linkID) + link, err := protonDrive.c.GetVolumeLink(ctx, protonDrive.MainShare.VolumeID, linkID) if err != nil { return nil, err } diff --git a/delete.go b/delete.go index 1ab1aaf..a1335d6 100644 --- a/delete.go +++ b/delete.go @@ -6,8 +6,8 @@ import ( "github.com/rclone/go-proton-api" ) -func (protonDrive *ProtonDrive) moveToTrash(ctx context.Context, parentLinkID string, linkIDs ...string) error { - err := protonDrive.c.TrashChildren(ctx, protonDrive.MainShare.ShareID, parentLinkID, linkIDs...) +func (protonDrive *ProtonDrive) moveToTrash(ctx context.Context, linkIDs ...string) error { + err := protonDrive.c.TrashVolumeLinks(ctx, protonDrive.MainShare.VolumeID, linkIDs...) if err != nil { return err } @@ -31,7 +31,7 @@ func (protonDrive *ProtonDrive) MoveFileToTrashByID(ctx context.Context, linkID return ErrLinkTypeMustToBeFileType } - return protonDrive.moveToTrash(ctx, fileLink.ParentLinkID, linkID) + return protonDrive.moveToTrash(ctx, linkID) } func (protonDrive *ProtonDrive) MoveFolderToTrashByID(ctx context.Context, linkID string, onlyOnEmpty bool) error { @@ -46,7 +46,7 @@ func (protonDrive *ProtonDrive) MoveFolderToTrashByID(ctx context.Context, linkI return ErrLinkTypeMustToBeFolderType } - childrenLinks, err := protonDrive.c.ListChildren(ctx, protonDrive.MainShare.ShareID, linkID /* false: list only active ones */, false) + childrenLinks, err := protonDrive.c.ListVolumeChildren(ctx, protonDrive.MainShare.VolumeID, linkID /* false: list only active ones */, false) if err != nil { return err } @@ -57,7 +57,7 @@ func (protonDrive *ProtonDrive) MoveFolderToTrashByID(ctx context.Context, linkI } } - return protonDrive.moveToTrash(ctx, folderLink.ParentLinkID, linkID) + return protonDrive.moveToTrash(ctx, linkID) } // WARNING!!!! @@ -66,7 +66,7 @@ func (protonDrive *ProtonDrive) MoveFolderToTrashByID(ctx context.Context, linkI func (protonDrive *ProtonDrive) EmptyRootFolder(ctx context.Context) error { protonDrive.ClearCache() - links, err := protonDrive.c.ListChildren(ctx, protonDrive.MainShare.ShareID, protonDrive.MainShare.LinkID, true) + links, err := protonDrive.c.ListVolumeChildren(ctx, protonDrive.MainShare.VolumeID, protonDrive.MainShare.LinkID, true) if err != nil { return err } @@ -79,7 +79,7 @@ func (protonDrive *ProtonDrive) EmptyRootFolder(ctx context.Context) error { } } - err := protonDrive.c.TrashChildren(ctx, protonDrive.MainShare.ShareID, protonDrive.MainShare.LinkID, linkIDs...) + err := protonDrive.c.TrashVolumeLinks(ctx, protonDrive.MainShare.VolumeID, linkIDs...) if err != nil { return err } @@ -93,7 +93,7 @@ func (protonDrive *ProtonDrive) EmptyRootFolder(ctx context.Context) error { } } - err := protonDrive.c.DeleteChildren(ctx, protonDrive.MainShare.ShareID, protonDrive.MainShare.LinkID, linkIDs...) + err := protonDrive.c.DeleteVolumeLinks(ctx, protonDrive.MainShare.VolumeID, linkIDs...) if err != nil { return err } @@ -106,7 +106,7 @@ func (protonDrive *ProtonDrive) EmptyRootFolder(ctx context.Context) error { func (protonDrive *ProtonDrive) EmptyTrash(ctx context.Context) error { protonDrive.ClearCache() - err := protonDrive.c.EmptyTrash(ctx, protonDrive.MainShare.ShareID) + err := protonDrive.c.EmptyVolumeTrash(ctx, protonDrive.MainShare.VolumeID) if err != nil { return err } diff --git a/drive.go b/drive.go index eeeac40..a186880 100644 --- a/drive.go +++ b/drive.go @@ -66,10 +66,12 @@ func NewProtonDrive(ctx context.Context, config *common.Config, authHandler prot // log.Printf("all volumes %#v", volumes) mainShareID := "" + mainVolumeID := "" for i := range volumes { // iOS drive: first active volume if volumes[i].State == proton.VolumeStateActive { mainShareID = volumes[i].Share.ShareID + mainVolumeID = volumes[i].VolumeID } } // log.Println("total volumes", len(volumes), "mainShareID", mainShareID) @@ -90,6 +92,7 @@ func NewProtonDrive(ctx context.Context, config *common.Config, authHandler prot for i := range shares { if shares[i].ShareID == mainShare.ShareID && shares[i].LinkID == mainShare.LinkID && + shares[i].VolumeID == mainVolumeID && shares[i].Flags == proton.PrimaryShare && shares[i].Type == proton.ShareTypeMain { mainShareCheck = true @@ -111,7 +114,7 @@ func NewProtonDrive(ctx context.Context, config *common.Config, authHandler prot Links also hold the file name (encrypted) and a hash of the name for name collisions. Link data is encrypted with its owning Share keyring. */ - rootLink, err := c.GetLink(ctx, mainShare.ShareID, mainShare.LinkID) + rootLink, err := c.GetVolumeLink(ctx, mainVolumeID, mainShare.LinkID) if err != nil { return nil, nil, err } diff --git a/file.go b/file.go index e5d6263..cab9954 100644 --- a/file.go +++ b/file.go @@ -17,7 +17,7 @@ type FileSystemAttrs struct { } func (protonDrive *ProtonDrive) GetRevisions(ctx context.Context, link *proton.Link, revisionType proton.RevisionState) ([]*proton.RevisionMetadata, error) { - revisions, err := protonDrive.c.ListRevisions(ctx, protonDrive.MainShare.ShareID, link.LinkID) + revisions, err := protonDrive.c.ListVolumeRevisions(ctx, protonDrive.MainShare.VolumeID, link.LinkID) if err != nil { return nil, err } @@ -104,7 +104,7 @@ func (protonDrive *ProtonDrive) GetActiveRevisionWithAttrs(ctx context.Context, return nil, nil, ErrCantFindActiveRevision } - revision, err := protonDrive.c.GetRevisionAllBlocks(ctx, protonDrive.MainShare.ShareID, link.LinkID, revisionsMetadata[0].ID) + revision, err := protonDrive.c.GetVolumeRevisionAllBlocks(ctx, protonDrive.MainShare.VolumeID, link.LinkID, revisionsMetadata[0].ID) if err != nil { return nil, nil, err } diff --git a/file_upload.go b/file_upload.go index be3d4d3..48c7e30 100644 --- a/file_upload.go +++ b/file_upload.go @@ -8,6 +8,8 @@ import ( "crypto/sha256" "encoding/base64" "encoding/hex" + "errors" + "fmt" "io" "mime" "os" @@ -19,6 +21,156 @@ import ( "github.com/rclone/go-proton-api" ) +const ( + blockUploadMaxAttempts = 5 + blockUploadRetryBaseDelay = time.Second + blockUploadRetryMaxDelay = 15 * time.Second +) + +type pendingUploadBlock struct { + blockUploadInfo proton.BlockUploadInfo + encData []byte +} + +type blockUploadResult struct { + index int + err error +} + +type blockUploadRetryLogger interface { + Warnf(format string, v ...interface{}) +} + +func retryableBlockUploadError(err error) bool { + if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + + var apiErr *proton.APIError + if errors.As(err, &apiErr) { + return apiErr.Status >= 500 && apiErr.Status <= 599 + } + + var protonNetErr *proton.NetError + return errors.As(err, &protonNetErr) +} + +func blockUploadRetryDelay(failedAttempt int) time.Duration { + delay := blockUploadRetryBaseDelay + for i := 1; i < failedAttempt && delay < blockUploadRetryMaxDelay; i++ { + delay *= 2 + } + if delay > blockUploadRetryMaxDelay { + return blockUploadRetryMaxDelay + } + return delay +} + +func waitForBlockUploadRetry(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func uploadBlockBatchWithRetry( + ctx context.Context, + blocks []pendingUploadBlock, + maxAttempts int, + requestLinks func(context.Context, []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error), + uploadBlock func(context.Context, proton.BlockUploadLink, []byte) error, + wait func(context.Context, time.Duration) error, + logger blockUploadRetryLogger, +) error { + remaining := append([]pendingUploadBlock(nil), blocks...) + var lastErr error + + for attempt := 1; attempt <= maxAttempts; attempt++ { + blockList := make([]proton.BlockUploadInfo, len(remaining)) + for i := range remaining { + blockList[i] = remaining[i].blockUploadInfo + } + + links, err := requestLinks(ctx, blockList) + if err != nil { + lastErr = err + if !retryableBlockUploadError(err) || attempt == maxAttempts { + return err + } + } else { + if len(links) != len(remaining) { + return fmt.Errorf( + "requested %d Proton block upload links, received %d", + len(remaining), + len(links), + ) + } + + results := make(chan blockUploadResult, len(remaining)) + for i := range remaining { + go func(index int) { + results <- blockUploadResult{ + index: index, + err: uploadBlock(ctx, links[index], remaining[index].encData), + } + }(i) + } + + errorsByIndex := make([]error, len(remaining)) + for range remaining { + result := <-results + errorsByIndex[result.index] = result.err + } + + failed := make([]pendingUploadBlock, 0, len(remaining)) + var terminalErr error + lastErr = nil + for i, uploadErr := range errorsByIndex { + if uploadErr == nil { + continue + } + if !retryableBlockUploadError(uploadErr) && terminalErr == nil { + terminalErr = uploadErr + } + if lastErr == nil { + lastErr = uploadErr + } + failed = append(failed, remaining[i]) + } + if terminalErr != nil { + return terminalErr + } + if len(failed) == 0 { + return nil + } + if attempt == maxAttempts { + return lastErr + } + remaining = failed + } + + delay := blockUploadRetryDelay(attempt) + if logger != nil { + logger.Warnf( + "Retrying %d transient Proton block upload(s) after %s (attempt %d/%d)", + len(remaining), + delay, + attempt+1, + maxAttempts, + ) + } + if err := wait(ctx, delay); err != nil { + return err + } + } + + return lastErr +} + func (protonDrive *ProtonDrive) handleRevisionConflict(ctx context.Context, link *proton.Link, createFileResp *proton.CreateFileRes) (string, bool, error) { if link != nil { linkID := link.LinkID @@ -38,7 +190,7 @@ func (protonDrive *ProtonDrive) handleRevisionConflict(ctx context.Context, link // delete the link (skipping trash, otherwise it won't work) and // signal the caller to resubmit the file creation request - err := protonDrive.c.DeleteChildren(ctx, protonDrive.MainShare.ShareID, link.ParentLinkID, linkID) + err := protonDrive.c.DeleteVolumeLinks(ctx, protonDrive.MainShare.VolumeID, linkID) if err != nil { return "", false, err } @@ -63,14 +215,14 @@ func (protonDrive *ProtonDrive) handleRevisionConflict(ctx context.Context, link // Question: how do we observe for file upload cancellation -> clientUID? // Random thoughts: if there are concurrent modification to the draft, the server should be able to catch this when commiting the revision // since the manifestSignature (hash) will fail to match - err = protonDrive.c.DeleteRevision(ctx, protonDrive.MainShare.ShareID, linkID, draftRevision[0].ID) + err = protonDrive.c.DeleteVolumeRevision(ctx, protonDrive.MainShare.VolumeID, linkID, draftRevision[0].ID) if err != nil { return "", false, err } } // create a new revision - newRevision, err := protonDrive.c.CreateRevision(ctx, protonDrive.MainShare.ShareID, linkID) + newRevision, err := protonDrive.c.CreateVolumeRevision(ctx, protonDrive.MainShare.VolumeID, linkID) if err != nil { return "", false, err } @@ -169,7 +321,7 @@ func (protonDrive *ProtonDrive) createFileUploadDraft(ctx context.Context, paren } createFileAction := func() (*proton.CreateFileRes, *proton.Link, error) { - createFileResp, err := protonDrive.c.CreateFile(ctx, protonDrive.MainShare.ShareID, createFileReq) + createFileResp, err := protonDrive.c.CreateVolumeFile(ctx, protonDrive.MainShare.VolumeID, createFileReq) if err != nil { // FIXME: check for duplicated filename by relying on checkAvailableHashes -> able to retrieve linkID too // Also saving generating resources such as new nodeKR, etc. @@ -257,62 +409,46 @@ func (protonDrive *ProtonDrive) createFileUploadDraft(ctx context.Context, paren } func (protonDrive *ProtonDrive) uploadAndCollectBlockData(ctx context.Context, newSessionKey *crypto.SessionKey, newNodeKR *crypto.KeyRing, file io.Reader, linkID, revisionID string) ([]byte, int64, []int64, string, error) { - type PendingUploadBlocks struct { - blockUploadInfo proton.BlockUploadInfo - encData []byte - } - if newSessionKey == nil || newNodeKR == nil { return nil, 0, nil, "", ErrMissingInputUploadAndCollectBlockData } totalFileSize := int64(0) - pendingUploadBlocks := make([]PendingUploadBlocks, 0) + pendingUploadBlocks := make([]pendingUploadBlock, 0) manifestSignatureData := make([]byte, 0) uploadPendingBlocks := func() error { if len(pendingUploadBlocks) == 0 { return nil } - blockList := make([]proton.BlockUploadInfo, 0) - for i := range pendingUploadBlocks { - blockList = append(blockList, pendingUploadBlocks[i].blockUploadInfo) - } - blockUploadReq := proton.BlockUploadReq{ - AddressID: protonDrive.MainShare.AddressID, - ShareID: protonDrive.MainShare.ShareID, - LinkID: linkID, - RevisionID: revisionID, - - BlockList: blockList, - } - blockUploadResp, err := protonDrive.c.RequestBlockUpload(ctx, blockUploadReq) - if err != nil { - return err + requestLinks := func(ctx context.Context, blockList []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + return protonDrive.c.RequestBlockUpload(ctx, proton.BlockUploadReq{ + AddressID: protonDrive.MainShare.AddressID, + VolumeID: protonDrive.MainShare.VolumeID, + LinkID: linkID, + RevisionID: revisionID, + BlockList: blockList, + }) } - - errChan := make(chan error) - uploadBlockWrapper := func(ctx context.Context, errChan chan error, bareURL, token string, block io.Reader) { - // log.Println("Before semaphore") + uploadBlock := func(ctx context.Context, link proton.BlockUploadLink, block []byte) error { if err := protonDrive.blockUploadSemaphore.Acquire(ctx, 1); err != nil { - errChan <- err + return err } defer protonDrive.blockUploadSemaphore.Release(1) - // log.Println("After semaphore") - // defer log.Println("Release semaphore") - errChan <- protonDrive.c.UploadBlock(ctx, bareURL, token, block) + return protonDrive.c.UploadBlock(ctx, link.BareURL, link.Token, bytes.NewReader(block)) } - for i := range blockUploadResp { - go uploadBlockWrapper(ctx, errChan, blockUploadResp[i].BareURL, blockUploadResp[i].Token, bytes.NewReader(pendingUploadBlocks[i].encData)) - } - - for i := 0; i < len(blockUploadResp); i++ { - err := <-errChan - if err != nil { - return err - } + if err := uploadBlockBatchWithRetry( + ctx, + pendingUploadBlocks, + blockUploadMaxAttempts, + requestLinks, + uploadBlock, + waitForBlockUploadRetry, + protonDrive.Config.GetLogger(), + ); err != nil { + return err } pendingUploadBlocks = pendingUploadBlocks[:0] @@ -405,7 +541,7 @@ func (protonDrive *ProtonDrive) uploadAndCollectBlockData(ctx context.Context, n } manifestSignatureData = append(manifestSignatureData, hash...) - pendingUploadBlocks = append(pendingUploadBlocks, PendingUploadBlocks{ + pendingUploadBlocks = append(pendingUploadBlocks, pendingUploadBlock{ blockUploadInfo: proton.BlockUploadInfo{ Index: i, // iOS drive: BE starts with 1 Size: int64(len(encData)), @@ -446,7 +582,7 @@ func (protonDrive *ProtonDrive) commitNewRevision(ctx context.Context, nodeKR *c return err } - err = protonDrive.c.CommitRevision(ctx, protonDrive.MainShare.ShareID, linkID, revisionID, commitRevisionReq) + err = protonDrive.c.CommitVolumeRevision(ctx, protonDrive.MainShare.VolumeID, linkID, revisionID, commitRevisionReq) if err != nil { return err } diff --git a/file_upload_concurrency_test.go b/file_upload_concurrency_test.go new file mode 100644 index 0000000..06d3677 --- /dev/null +++ b/file_upload_concurrency_test.go @@ -0,0 +1,235 @@ +package proton_api_bridge + +import ( + "context" + "errors" + "reflect" + "sync" + "testing" + "time" + + "github.com/rclone/go-proton-api" + "golang.org/x/sync/semaphore" +) + +func testPendingBlocks(indexes ...int) []pendingUploadBlock { + blocks := make([]pendingUploadBlock, len(indexes)) + for i, index := range indexes { + blocks[i] = pendingUploadBlock{ + blockUploadInfo: proton.BlockUploadInfo{Index: index}, + encData: []byte{byte(index)}, + } + } + return blocks +} + +func testUploadLinks(blocks []proton.BlockUploadInfo) []proton.BlockUploadLink { + links := make([]proton.BlockUploadLink, len(blocks)) + for i, block := range blocks { + links[i] = proton.BlockUploadLink{Token: string(rune(block.Index))} + } + return links +} + +func noRetryWait(_ context.Context, _ time.Duration) error { return nil } + +func TestUploadBlockBatchRetriesOnlyTransientFailures(t *testing.T) { + var requestIndexes [][]int + requestLinks := func(_ context.Context, blocks []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + indexes := make([]int, len(blocks)) + for i, block := range blocks { + indexes[i] = block.Index + } + requestIndexes = append(requestIndexes, indexes) + return testUploadLinks(blocks), nil + } + + var mu sync.Mutex + uploads := map[int]int{} + uploadBlock := func(_ context.Context, _ proton.BlockUploadLink, block []byte) error { + index := int(block[0]) + mu.Lock() + uploads[index]++ + attempt := uploads[index] + mu.Unlock() + if attempt == 1 && index != 2 { + return &proton.APIError{Status: 502, Message: "temporary storage failure"} + } + return nil + } + + err := uploadBlockBatchWithRetry( + context.Background(), + testPendingBlocks(1, 2, 3), + blockUploadMaxAttempts, + requestLinks, + uploadBlock, + noRetryWait, + nil, + ) + if err != nil { + t.Fatalf("retrying transient block uploads failed: %v", err) + } + if want := [][]int{{1, 2, 3}, {1, 3}}; !reflect.DeepEqual(requestIndexes, want) { + t.Fatalf("requested block indexes %v, want %v", requestIndexes, want) + } + if want := map[int]int{1: 2, 2: 1, 3: 2}; !reflect.DeepEqual(uploads, want) { + t.Fatalf("block upload counts %v, want %v", uploads, want) + } +} + +func TestUploadBlockBatchReturnsNonRetryableError(t *testing.T) { + terminalErr := &proton.APIError{Status: 422, Message: "draft conflict"} + requests := 0 + err := uploadBlockBatchWithRetry( + context.Background(), + testPendingBlocks(1), + blockUploadMaxAttempts, + func(_ context.Context, blocks []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + requests++ + return testUploadLinks(blocks), nil + }, + func(_ context.Context, _ proton.BlockUploadLink, _ []byte) error { return terminalErr }, + noRetryWait, + nil, + ) + if !errors.Is(err, terminalErr) { + t.Fatalf("returned error %v, want %v", err, terminalErr) + } + if requests != 1 { + t.Fatalf("requested upload links %d times after a terminal error, want 1", requests) + } +} + +func TestUploadBlockBatchRetriesTransientLinkRequest(t *testing.T) { + transientErr := &proton.APIError{Status: 502, Message: "temporary API failure"} + requests := 0 + uploads := 0 + err := uploadBlockBatchWithRetry( + context.Background(), + testPendingBlocks(1), + blockUploadMaxAttempts, + func(_ context.Context, blocks []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + requests++ + if requests == 1 { + return nil, transientErr + } + return testUploadLinks(blocks), nil + }, + func(_ context.Context, _ proton.BlockUploadLink, _ []byte) error { + uploads++ + return nil + }, + noRetryWait, + nil, + ) + if err != nil { + t.Fatalf("retrying a transient link request failed: %v", err) + } + if requests != 2 || uploads != 1 { + t.Fatalf("observed %d link requests and %d uploads, want 2 and 1", requests, uploads) + } +} + +func TestUploadBlockBatchRejectsMismatchedLinkCount(t *testing.T) { + err := uploadBlockBatchWithRetry( + context.Background(), + testPendingBlocks(1, 2), + blockUploadMaxAttempts, + func(_ context.Context, _ []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + return []proton.BlockUploadLink{{}}, nil + }, + func(_ context.Context, _ proton.BlockUploadLink, _ []byte) error { + t.Fatal("upload must not start with a mismatched link response") + return nil + }, + noRetryWait, + nil, + ) + if err == nil { + t.Fatal("expected a mismatched link count to fail") + } +} + +func TestUploadBlockBatchReturnsLastErrorAfterLimit(t *testing.T) { + transientErr := &proton.APIError{Status: 502, Message: "temporary storage failure"} + requests := 0 + err := uploadBlockBatchWithRetry( + context.Background(), + testPendingBlocks(1), + 3, + func(_ context.Context, blocks []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + requests++ + return testUploadLinks(blocks), nil + }, + func(_ context.Context, _ proton.BlockUploadLink, _ []byte) error { return transientErr }, + noRetryWait, + nil, + ) + if !errors.Is(err, transientErr) { + t.Fatalf("returned error %v, want %v", err, transientErr) + } + if requests != 3 { + t.Fatalf("requested upload links %d times, want 3", requests) + } +} + +func TestUploadBlockBatchHonorsCancellationDuringBackoff(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := uploadBlockBatchWithRetry( + ctx, + testPendingBlocks(1), + blockUploadMaxAttempts, + func(_ context.Context, blocks []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + return testUploadLinks(blocks), nil + }, + func(_ context.Context, _ proton.BlockUploadLink, _ []byte) error { + return &proton.APIError{Status: 502, Message: "temporary storage failure"} + }, + waitForBlockUploadRetry, + nil, + ) + if !errors.Is(err, context.Canceled) { + t.Fatalf("returned error %v, want context cancellation", err) + } +} + +func TestUploadBlockBatchReleasesAllWorkersAfterFailure(t *testing.T) { + const slotCount = int64(20) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + slots := semaphore.NewWeighted(slotCount) + + for batch := 0; batch < 4; batch++ { + err := uploadBlockBatchWithRetry( + ctx, + testPendingBlocks(1, 2, 3, 4, 5, 6, 7, 8), + blockUploadMaxAttempts, + func(_ context.Context, blocks []proton.BlockUploadInfo) ([]proton.BlockUploadLink, error) { + return testUploadLinks(blocks), nil + }, + func(_ context.Context, _ proton.BlockUploadLink, block []byte) error { + if err := slots.Acquire(ctx, 1); err != nil { + return err + } + defer slots.Release(1) + if block[0] == 1 { + return errors.New("synthetic upload failure") + } + return nil + }, + noRetryWait, + nil, + ) + if err == nil { + t.Fatal("expected the first upload failure to be returned") + } + } + + if err := slots.Acquire(ctx, slotCount); err != nil { + t.Fatalf("upload workers leaked semaphore slots: %v", err) + } + slots.Release(slotCount) +} diff --git a/folder.go b/folder.go index 17d5879..971380a 100644 --- a/folder.go +++ b/folder.go @@ -24,7 +24,7 @@ func (protonDrive *ProtonDrive) ListDirectory( } if folderLink.State == proton.LinkStateActive { - childrenLinks, err := protonDrive.c.ListChildren(ctx, protonDrive.MainShare.ShareID, folderLink.LinkID, true) + childrenLinks, err := protonDrive.c.ListVolumeChildren(ctx, protonDrive.MainShare.VolumeID, folderLink.LinkID, true) if err != nil { return nil, err } @@ -137,7 +137,7 @@ func (protonDrive *ProtonDrive) CreateNewFolder(ctx context.Context, parentLink // FIXME: check for duplicated filename by relying on checkAvailableHashes // if the folder name already exist, this call will return an error - createFolderResp, err := protonDrive.c.CreateFolder(ctx, protonDrive.MainShare.ShareID, createFolderReq) + createFolderResp, err := protonDrive.c.CreateVolumeFolder(ctx, protonDrive.MainShare.VolumeID, createFolderReq) if err != nil { return "", err } @@ -251,7 +251,7 @@ func (protonDrive *ProtonDrive) moveLink(ctx context.Context, srcLink *proton.Li // TODO: disable cache when move is in action? // because there might be the case where others read for the same link currently being move -> race condition // argument: cache itself is already outdated in a sense, as we don't even have event system (even if we have, it's still outdated...) - err = protonDrive.c.MoveLink(ctx, protonDrive.MainShare.ShareID, srcLink.LinkID, req) + err = protonDrive.c.MoveVolumeLink(ctx, protonDrive.MainShare.VolumeID, srcLink.LinkID, req) if err != nil { return err } diff --git a/folder_recursive.go b/folder_recursive.go index 5f3bc69..dc21366 100644 --- a/folder_recursive.go +++ b/folder_recursive.go @@ -84,7 +84,7 @@ func (protonDrive *ProtonDrive) listDirectoriesRecursively( if maxDepth == -1 || curDepth < maxDepth { if link.Type == proton.LinkTypeFolder { - childrenLinks, err := protonDrive.c.ListChildren(ctx, protonDrive.MainShare.ShareID, link.LinkID, true) + childrenLinks, err := protonDrive.c.ListVolumeChildren(ctx, protonDrive.MainShare.VolumeID, link.LinkID, true) if err != nil { return err } diff --git a/go.mod b/go.mod index 30a5df7..aad02b6 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,8 @@ go 1.23.0 toolchain go1.23.5 +replace github.com/rclone/go-proton-api => github.com/oss-singularity/go-proton-api v1.0.5-0.20260916221756-b491f8604bbd + require ( github.com/ProtonMail/gluon v0.17.1-0.20230724134000-308be39be96e github.com/ProtonMail/go-crypto v1.4.1 diff --git a/go.sum b/go.sum index e249d22..f9299ac 100644 --- a/go.sum +++ b/go.sum @@ -67,14 +67,14 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/oss-singularity/go-proton-api v1.0.5-0.20260916221756-b491f8604bbd h1:YCWhbaHk2k31mKPQCwCsI3E2LKsEZ8frv57D9FhGVcQ= +github.com/oss-singularity/go-proton-api v1.0.5-0.20260916221756-b491f8604bbd/go.mod h1:QAlkFfswzrBuxvCORWV8rZdddg52hahMN98CFWoFW1E= github.com/pelletier/go-toml/v2 v2.0.8 h1:0ctb6s9mE31h0/lhu+J6OPmVeDxJn+kYnJc2jZR9tGQ= github.com/pelletier/go-toml/v2 v2.0.8/go.mod h1:vuYfssBdrU2XDZ9bYydBu6t+6a6PYNcZljzZR9VXg+4= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/rclone/go-proton-api v1.0.4 h1:AJW0e9pB4j0hVK4WqyGErFwaI+5MUQWPCtj5FYYxtPg= -github.com/rclone/go-proton-api v1.0.4/go.mod h1:QAlkFfswzrBuxvCORWV8rZdddg52hahMN98CFWoFW1E= github.com/relvacode/iso8601 v1.6.0 h1:eFXUhMJN3Gz8Rcq82f9DTMW0svjtAVuIEULglM7QHTU= github.com/relvacode/iso8601 v1.6.0/go.mod h1:FlNp+jz+TXpyRqgmM7tnzHHzBnz776kmAH2h3sZCn0I= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= diff --git a/search.go b/search.go index a3560cc..204990e 100644 --- a/search.go +++ b/search.go @@ -77,7 +77,7 @@ func (protonDrive *ProtonDrive) SearchByNameInActiveFolder( // use available hash to check if it exists // more efficient than linear scan to just do existence check // used in rclone when Put(), it will try to see if the object exists or not - res, err := protonDrive.c.CheckAvailableHashes(ctx, protonDrive.MainShare.ShareID, folderLink.LinkID, proton.CheckAvailableHashesReq{ + res, err := protonDrive.c.CheckVolumeAvailableHashes(ctx, protonDrive.MainShare.VolumeID, folderLink.LinkID, proton.CheckAvailableHashesReq{ Hashes: []string{targetNameHash}, }) if err != nil { @@ -89,7 +89,7 @@ func (protonDrive *ProtonDrive) SearchByNameInActiveFolder( return nil, nil } - childrenLinks, err := protonDrive.c.ListChildren(ctx, protonDrive.MainShare.ShareID, folderLink.LinkID, true) + childrenLinks, err := protonDrive.c.ListVolumeChildren(ctx, protonDrive.MainShare.VolumeID, folderLink.LinkID, true) if err != nil { return nil, err } diff --git a/search_recursive.go b/search_recursive.go index 5b23100..7c368a4 100644 --- a/search_recursive.go +++ b/search_recursive.go @@ -87,7 +87,7 @@ func (protonDrive *ProtonDrive) performSearchByNameRecursively( } if link.Type == proton.LinkTypeFolder { - childrenLinks, err := protonDrive.c.ListChildren(ctx, protonDrive.MainShare.ShareID, link.LinkID, true) + childrenLinks, err := protonDrive.c.ListVolumeChildren(ctx, protonDrive.MainShare.VolumeID, link.LinkID, true) if err != nil { return nil, err }