From 4ae08ef88ead3af7e317d73b925f21702fee3d0d Mon Sep 17 00:00:00 2001 From: mirivlad Date: Fri, 17 Jul 2026 05:09:41 +0800 Subject: [PATCH] Use paginated blob transport for file sync --- README.md | 25 +++- docs/PLUGIN_RUNTIME.md | 14 +- internal/api/app.go | 183 ++++++++++++++++------- internal/api/app_test.go | 48 +++++- internal/api/sync_real_server_test.go | 21 +++ internal/core/files/service.go | 80 ++++++++++ internal/core/sync/client.go | 207 +++++++++++++++++++++----- internal/core/sync/client_test.go | 97 ++++++++++++ internal/core/sync/snapshot.go | 117 +++++++++++---- internal/core/sync/snapshot_test.go | 41 ++++- scripts/smoke-real-sync.sh | 8 +- 11 files changed, 708 insertions(+), 133 deletions(-) diff --git a/README.md b/README.md index 7766c07..977d3e1 100644 --- a/README.md +++ b/README.md @@ -233,14 +233,23 @@ deletes, and incompatible files become visible conflicts rather than silent overwrites. Pairing can optionally use an existing remote vault ID to restore that scope on a new device. -The current operation transport supports ordinary files and folders plus -workspace (Deal) create/rename/trash/restore with a durable workspace UUID. -`.verstak`, trash, temporary files, and symlinks are excluded from ordinary -file sync. Text remains bounded to 2 MB and binary payloads to 8 MB; a larger -or unsupported file is left as a visible unresolved sync warning instead of -being treated as synchronized. Blob transfer, quotas, pagination, retention, -and Secrets, plugin settings, Todo, Journal, Activity, and Browser Inbox sync -are future milestones. +The current transport supports ordinary files and folders plus workspace (Deal) +create/rename/trash/restore with a durable workspace UUID. `.verstak`, trash, +temporary files, and symlinks are excluded from ordinary file sync. UTF-8 text +remains inline only within the existing 2 MB bound; binary and larger content +is staged privately under `.verstak/sync/blobs`, uploaded through the scoped +Blob API, and represented in the operation log only by SHA-256 and size. The +download is streamed, hash/size-verified, and atomically applied. A file over +the configured server blob limit or otherwise unsupported remains a visible +unresolved warning instead of being treated as synchronized. + +Pull is paginated. Desktop applies operations in increasing server sequence, +persists its cursor only after each successful operation, stops before later +pages at the first failure, and retries after restart. Blob references are +authorized per user/vault; a missing or corrupt blob likewise leaves the cursor +unchanged. The server is optional and file bytes are not end-to-end encrypted. +Operation-log retention/checkpoints and synchronization of Secrets, plugin +settings, Todo, Journal, Activity, and Browser Inbox remain future milestones. ## Build from source diff --git a/docs/PLUGIN_RUNTIME.md b/docs/PLUGIN_RUNTIME.md index 2f96314..cbebef8 100644 --- a/docs/PLUGIN_RUNTIME.md +++ b/docs/PLUGIN_RUNTIME.md @@ -511,12 +511,14 @@ contributions summary. lifecycle is a separate core `workspace` entity (`create`, `rename`, `trash`, `restore`) carrying the durable `workspaceId`; plugins never obtain access to the nested workspace marker. -- File transport remains bounded: UTF-8 text uses the 2 MB Files API limit and - binary/other regular files use the existing base64 path up to 8 MB. Larger or - unsupported files remain visible unresolved warnings and are not added to a - successful snapshot. Blob transport, quotas, pagination, retention, and - synchronization of Secrets, plugin settings, Todo, Journal, Activity, and - Browser Inbox are future work. +- File transport remains bounded: UTF-8 text uses the 2 MB Files API limit; + binary and larger regular files are staged privately, sent through Blob API, + and referenced by SHA-256/size rather than base64 in `payload_json`. Desktop + streams and verifies downloads before atomic apply. Pull is paginated and the + durable cursor advances only after every applied sequence. Files above the + configured blob limit or otherwise unsupported remain unresolved warnings. + Operation retention/checkpoints and synchronization of Secrets, plugin + settings, Todo, Journal, Activity, and Browser Inbox are future work. - Transport push/pull uses bounded retry/backoff for transient HTTP/network failures. Client/auth errors are not retried. diff --git a/internal/api/app.go b/internal/api/app.go index b5de9e5..f8f9956 100644 --- a/internal/api/app.go +++ b/internal/api/app.go @@ -3459,6 +3459,15 @@ func (a *App) syncNow() (map[string]interface{}, error) { } pushResult := &syncsvc.PushResponse{} if len(unpushed) > 0 { + if err := a.uploadPendingBlobs(client, unpushed); err != nil { + message := fmt.Sprintf("blob upload: %v", err) + _ = a.updateSyncError(message) + // A server-side file/quota limit is unresolved scanner input from the + // user's perspective: retain it visibly and keep the operation pending + // so a later sync retries rather than silently accepting the snapshot. + _ = a.syncSvc.SetLastWarning(message) + return nil, fmt.Errorf("blob upload: %w", err) + } pushResult, err = client.Push(unpushed) if err != nil { _ = a.updateSyncError(fmt.Sprintf("push: %v", err)) @@ -3505,58 +3514,96 @@ func (a *App) syncNow() (map[string]interface{}, error) { return result, nil } -func (a *App) pullRemoteOps(client *syncsvc.Client, cursor int, initialReconciliation bool) (pulled, nextCursor, serverSequence int, err error) { - pullResult, err := client.Pull(cursor) - if err != nil { - _ = a.updateSyncError(fmt.Sprintf("pull: %v", err)) - return 0, cursor, cursor, fmt.Errorf("pull: %w", err) - } - nextCursor = cursor - lastSequenceInBatch := cursor - for _, op := range pullResult.Ops { - if op.ServerSequence <= cursor { +// uploadPendingBlobs ensures every operation referring to binary content has +// its immutable local cache uploaded before the operation reaches the log. +func (a *App) uploadPendingBlobs(client *syncsvc.Client, ops []syncsvc.Op) error { + for _, op := range ops { + if op.EntityType != syncsvc.EntityFile || (op.OpType != syncsvc.OpCreate && op.OpType != syncsvc.OpUpdate) { continue } - if op.ServerSequence <= lastSequenceInBatch { - errMsg := fmt.Sprintf("pull response is not strictly ordered at sequence %d (%s)", op.ServerSequence, op.OpID) - _ = a.updateSyncError(errMsg) - return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg) + payload, err := parseSyncFilePayload(op.PayloadJSON) + if err != nil { + return fmt.Errorf("parse operation %s: %w", op.OpID, err) } - lastSequenceInBatch = op.ServerSequence - if err := a.applyRemoteOpForReconciliation(op, initialReconciliation); err != nil { - path := syncOperationPath(op) - errMsg := fmt.Sprintf("pull apply failed at sequence %d for %s %s (%s): %v", op.ServerSequence, op.EntityType, path, op.OpID, err) - _ = a.updateSyncError(errMsg) - return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg) + if payload.Blob == nil { + continue } - if err := a.syncSvc.RecordRemoteOp(op); err != nil { - errMsg := fmt.Sprintf("record applied remote operation at sequence %d (%s): %v", op.ServerSequence, op.OpID, err) - _ = a.updateSyncError(errMsg) - return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg) + if payload.Blob.Size < 0 || payload.Blob.SHA256 == "" { + return fmt.Errorf("invalid blob reference in operation %s", op.OpID) } - if err := a.syncSvc.SetLastPullSeq(op.ServerSequence); err != nil { - errMsg := fmt.Sprintf("save pull cursor at sequence %d (%s): %v", op.ServerSequence, op.OpID, err) - _ = a.updateSyncError(errMsg) - return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg) + cachePath := syncsvc.BlobCachePath(a.vaultPath(), payload.Blob.SHA256) + ref, err := client.UploadBlob(cachePath) + if err != nil { + return fmt.Errorf("%s: %w", syncPayloadPath(op, payload), err) } - if warnings, err := a.syncSvc.RebaseSnapshot(); err != nil { - errMsg := fmt.Sprintf("rebase snapshot after sequence %d (%s): %v", op.ServerSequence, op.OpID, err) - _ = a.updateSyncError(errMsg) - return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("%s", errMsg) - } else if err := a.syncSvc.SetLastWarning(strings.Join(warnings, "\n")); err != nil { - return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("save sync warning: %w", err) + if ref.SHA256 != payload.Blob.SHA256 || ref.Size != payload.Blob.Size { + return fmt.Errorf("%s: uploaded blob does not match operation reference", syncPayloadPath(op, payload)) } - nextCursor = op.ServerSequence - pulled++ } - if pullResult.ServerSequence > nextCursor { - if err := a.syncSvc.SetLastPullSeq(pullResult.ServerSequence); err != nil { - _ = a.updateSyncError(fmt.Sprintf("save pull cursor at sequence %d: %v", pullResult.ServerSequence, err)) - return pulled, nextCursor, pullResult.ServerSequence, fmt.Errorf("save pull cursor: %w", err) + return nil +} + +func (a *App) pullRemoteOps(client *syncsvc.Client, cursor int, initialReconciliation bool) (pulled, nextCursor, serverSequence int, err error) { + nextCursor = cursor + for { + pageStartCursor := nextCursor + pullResult, pullErr := client.PullPage(nextCursor, 0) + if pullErr != nil { + _ = a.updateSyncError(fmt.Sprintf("pull: %v", pullErr)) + return pulled, nextCursor, serverSequence, fmt.Errorf("pull: %w", pullErr) + } + serverSequence = pullResult.ServerSequence + lastSequenceInPage := nextCursor + for _, op := range pullResult.Ops { + if op.ServerSequence <= nextCursor { + continue + } + if op.ServerSequence <= lastSequenceInPage { + errMsg := fmt.Sprintf("pull response is not strictly ordered at sequence %d (%s)", op.ServerSequence, op.OpID) + _ = a.updateSyncError(errMsg) + return pulled, nextCursor, serverSequence, fmt.Errorf("%s", errMsg) + } + lastSequenceInPage = op.ServerSequence + if err := a.applyRemoteOpWithClient(client, op, initialReconciliation); err != nil { + path := syncOperationPath(op) + errMsg := fmt.Sprintf("pull apply failed at sequence %d for %s %s (%s): %v", op.ServerSequence, op.EntityType, path, op.OpID, err) + _ = a.updateSyncError(errMsg) + return pulled, nextCursor, serverSequence, fmt.Errorf("%s", errMsg) + } + if err := a.syncSvc.RecordRemoteOp(op); err != nil { + errMsg := fmt.Sprintf("record applied remote operation at sequence %d (%s): %v", op.ServerSequence, op.OpID, err) + _ = a.updateSyncError(errMsg) + return pulled, nextCursor, serverSequence, fmt.Errorf("%s", errMsg) + } + if err := a.syncSvc.SetLastPullSeq(op.ServerSequence); err != nil { + errMsg := fmt.Sprintf("save pull cursor at sequence %d (%s): %v", op.ServerSequence, op.OpID, err) + _ = a.updateSyncError(errMsg) + return pulled, nextCursor, serverSequence, fmt.Errorf("%s", errMsg) + } + if warnings, err := a.syncSvc.RebaseSnapshot(); err != nil { + errMsg := fmt.Sprintf("rebase snapshot after sequence %d (%s): %v", op.ServerSequence, op.OpID, err) + _ = a.updateSyncError(errMsg) + return pulled, nextCursor, serverSequence, fmt.Errorf("%s", errMsg) + } else if err := a.syncSvc.SetLastWarning(strings.Join(warnings, "\n")); err != nil { + return pulled, nextCursor, serverSequence, fmt.Errorf("save sync warning: %w", err) + } + nextCursor = op.ServerSequence + pulled++ + } + if pullResult.PageLastSequence != lastSequenceInPage { + errMsg := fmt.Sprintf("pull page cursor mismatch: got %d, applied %d", pullResult.PageLastSequence, lastSequenceInPage) + _ = a.updateSyncError(errMsg) + return pulled, nextCursor, serverSequence, fmt.Errorf("%s", errMsg) + } + if !pullResult.HasMore { + return pulled, nextCursor, serverSequence, nil + } + if pullResult.PageLastSequence <= pageStartCursor { + errMsg := "pull page declared more operations without advancing cursor" + _ = a.updateSyncError(errMsg) + return pulled, nextCursor, serverSequence, fmt.Errorf("%s", errMsg) } - nextCursor = pullResult.ServerSequence } - return pulled, nextCursor, pullResult.ServerSequence, nil } func syncOperationPath(op syncsvc.Op) string { @@ -3601,10 +3648,14 @@ func (a *App) updateSyncSuccess(lastSyncAt string) error { } func (a *App) applyRemoteOp(op syncsvc.Op) error { - return a.applyRemoteOpForReconciliation(op, false) + return a.applyRemoteOpWithClient(nil, op, false) } func (a *App) applyRemoteOpForReconciliation(op syncsvc.Op, initialReconciliation bool) error { + return a.applyRemoteOpWithClient(nil, op, initialReconciliation) +} + +func (a *App) applyRemoteOpWithClient(client *syncsvc.Client, op syncsvc.Op, initialReconciliation bool) error { if a.debug { log.Printf("[sync] applyRemoteOp: type=%s entity=%s/%s", op.OpType, op.EntityType, op.EntityID) } @@ -3628,7 +3679,7 @@ func (a *App) applyRemoteOpForReconciliation(op syncsvc.Op, initialReconciliatio } switch op.EntityType { case syncsvc.EntityFile: - return a.applyRemoteFileOp(op, payload, initialReconciliation) + return a.applyRemoteFileOp(client, op, payload, initialReconciliation) case syncsvc.EntityFolder: return a.applyRemoteFolderOp(op, payload, initialReconciliation) default: @@ -3637,12 +3688,13 @@ func (a *App) applyRemoteOpForReconciliation(op syncsvc.Op, initialReconciliatio } type syncFilePayload struct { - Path string `json:"path"` - Content string `json:"content"` - DataBase64 *string `json:"dataBase64"` - ContentHash string `json:"contentHash"` - FromPath string `json:"fromPath"` - ToPath string `json:"toPath"` + Path string `json:"path"` + Content string `json:"content"` + DataBase64 *string `json:"dataBase64"` + Blob *syncsvc.BlobReference `json:"blob"` + ContentHash string `json:"contentHash"` + FromPath string `json:"fromPath"` + ToPath string `json:"toPath"` } type syncWorkspacePayload struct { @@ -3705,7 +3757,7 @@ func (a *App) applyRemoteWorkspaceOp(op syncsvc.Op, payload syncWorkspacePayload } } -func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload, initialReconciliation bool) error { +func (a *App) applyRemoteFileOp(client *syncsvc.Client, op syncsvc.Op, payload syncFilePayload, initialReconciliation bool) error { switch op.OpType { case syncsvc.OpCreate: path := syncPayloadPath(op, payload) @@ -3722,6 +3774,9 @@ func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload, initialR if exists { return fmt.Errorf("conflict: remote create would replace local file %s", path) } + if payload.Blob != nil { + return a.applyRemoteBlobFile(client, path, payload, corefiles.WriteOptions{CreateIfMissing: true}) + } if payload.DataBase64 != nil { return a.files.WriteVaultFileBytes(path, *payload.DataBase64, corefiles.WriteOptions{CreateIfMissing: true}) } @@ -3746,6 +3801,9 @@ func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload, initialR } else if pending { return fmt.Errorf("conflict: remote update would replace unpushed local file %s", path) } + if payload.Blob != nil { + return a.applyRemoteBlobFile(client, path, payload, corefiles.WriteOptions{CreateIfMissing: !exists, Overwrite: exists}) + } if payload.DataBase64 != nil { return a.files.WriteVaultFileBytes(path, *payload.DataBase64, corefiles.WriteOptions{CreateIfMissing: !exists, Overwrite: exists}) } @@ -3810,6 +3868,26 @@ func (a *App) applyRemoteFileOp(op syncsvc.Op, payload syncFilePayload, initialR } } +func (a *App) applyRemoteBlobFile(client *syncsvc.Client, path string, payload syncFilePayload, options corefiles.WriteOptions) error { + if client == nil { + return fmt.Errorf("blob operation requires an active sync client") + } + if payload.Blob == nil || payload.Blob.Size < 0 || payload.Blob.SHA256 == "" { + return fmt.Errorf("invalid remote blob reference") + } + if payload.ContentHash != "" && payload.ContentHash != payload.Blob.SHA256 { + return fmt.Errorf("remote blob hash does not match file content hash") + } + cachePath := syncsvc.BlobCachePath(a.vaultPath(), payload.Blob.SHA256) + if err := client.DownloadBlobVerified(payload.Blob.SHA256, payload.Blob.Size, cachePath); err != nil { + return fmt.Errorf("download blob: %w", err) + } + if err := a.files.WriteVaultFileFromPath(path, cachePath, options); err != nil { + return err + } + return nil +} + func (a *App) remoteFileMatches(path string, payload syncFilePayload) (matches, exists bool, err error) { normalized, err := corefiles.NormalizeRelativeFile(path) if err != nil { @@ -3845,6 +3923,9 @@ func remotePayloadHash(payload syncFilePayload) (string, error) { if payload.ContentHash != "" { return payload.ContentHash, nil } + if payload.Blob != nil { + return payload.Blob.SHA256, nil + } data := []byte(payload.Content) if payload.DataBase64 != nil { decoded, err := base64.StdEncoding.DecodeString(*payload.DataBase64) diff --git a/internal/api/app_test.go b/internal/api/app_test.go index 7016eeb..1a83256 100644 --- a/internal/api/app_test.go +++ b/internal/api/app_test.go @@ -1794,7 +1794,7 @@ func TestFileBridgeRecordsSyncOps(t *testing.T) { {syncsvc.EntityFolder, "Docs", syncsvc.OpCreate, `"path":"Docs"`}, {syncsvc.EntityFile, "Docs/one.txt", syncsvc.OpCreate, `"content":"hello"`}, {syncsvc.EntityFile, "Docs/one.txt", syncsvc.OpUpdate, `"content":"updated"`}, - {syncsvc.EntityFile, "Docs/image.bin", syncsvc.OpCreate, `"dataBase64":"AQID"`}, + {syncsvc.EntityFile, "Docs/image.bin", syncsvc.OpCreate, `"blob":{"sha256":`}, // Snapshot reconciliation represents external and API renames uniformly // as a create at the destination followed by a delete at the source. {syncsvc.EntityFile, "Docs/two.txt", syncsvc.OpCreate, `"content":"updated"`}, @@ -1948,6 +1948,52 @@ func TestSyncNowPushesLocalOpsAndAppliesPulledFileOps(t *testing.T) { } } +func TestSyncNowAppliesEveryPullPageInOrder(t *testing.T) { + app, root := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device") + server := newLocalHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer device-token" { + http.Error(w, "missing auth", http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/api/v1/sync/pull": + var request syncsvc.PullRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Fatal(err) + } + pages := map[int]map[string]interface{}{ + 0: {"server_sequence": 3, "page_last_sequence": 1, "has_more": true, "ops": []map[string]interface{}{{"op_id": "folder", "server_sequence": 1, "device_id": "remote", "entity_type": syncsvc.EntityFolder, "entity_id": "Remote", "op_type": syncsvc.OpCreate, "payload_json": `{"path":"Remote"}`}}}, + 1: {"server_sequence": 3, "page_last_sequence": 2, "has_more": true, "ops": []map[string]interface{}{{"op_id": "one", "server_sequence": 2, "device_id": "remote", "entity_type": syncsvc.EntityFile, "entity_id": "Remote/one.txt", "op_type": syncsvc.OpCreate, "payload_json": `{"path":"Remote/one.txt","content":"one"}`}}}, + 2: {"server_sequence": 3, "page_last_sequence": 3, "has_more": false, "ops": []map[string]interface{}{{"op_id": "two", "server_sequence": 3, "device_id": "remote", "entity_type": syncsvc.EntityFile, "entity_id": "Remote/two.txt", "op_type": syncsvc.OpCreate, "payload_json": `{"path":"Remote/two.txt","content":"two"}`}}}, + 3: {"server_sequence": 3, "page_last_sequence": 3, "has_more": false, "ops": []map[string]interface{}{}}, + } + _ = json.NewEncoder(w).Encode(pages[request.SinceSequence]) + case "/api/v1/sync/push": + _ = json.NewEncoder(w).Encode(map[string]interface{}{"accepted": []string{}, "count": 0, "conflicts": []map[string]interface{}{}}) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + if err := app.syncSvc.SetState(server.URL, ""); err != nil { + t.Fatal(err) + } + if err := syncsvc.SaveDeviceToken(root, "device-token"); err != nil { + t.Fatal(err) + } + result, err := app.syncNow() + if err != nil { + t.Fatal(err) + } + if result["pulled"] != 3 { + t.Fatalf("pull result = %#v, want three operations", result) + } + expectText(t, app, "Remote/one.txt", "one") + expectText(t, app, "Remote/two.txt", "two") + assertLastPullSequence(t, app.syncSvc, 3) +} + func TestSyncNowStopsAtFailedRemoteOperationAndRetriesAfterRestart(t *testing.T) { app, root := newSyncFilesTestApp(t, []string{"files.read", "files.write", "files.delete"}, "local-device") pullCalls := 0 diff --git a/internal/api/sync_real_server_test.go b/internal/api/sync_real_server_test.go index 1aa5060..7caf747 100644 --- a/internal/api/sync_real_server_test.go +++ b/internal/api/sync_real_server_test.go @@ -1,6 +1,7 @@ package api import ( + "bytes" "os" "path/filepath" "strings" @@ -104,6 +105,26 @@ func TestSyncNowAgainstRealServerTwoVaults(t *testing.T) { expectText(t, appB, "Shared/external.txt", "external while running") assertNoUnpushedOps(t, appB) + // This exceeds the former 8 MiB inline/base64 ceiling. The operation must + // contain only a blob reference; the actual bytes travel through Blob API. + binary := make([]byte, corefiles.MaxBinaryReadBytes+1) + for i := range binary { + binary[i] = byte(i % 251) + } + if err := os.WriteFile(filepath.Join(rootA, "Shared", "large.bin"), binary, 0o644); err != nil { + t.Fatalf("write large binary: %v", err) + } + expectSyncCounts(t, appA, 1, 1) + expectSyncCounts(t, appB, 0, 1) + received, err := os.ReadFile(filepath.Join(rootB, "Shared", "large.bin")) + if err != nil { + t.Fatalf("read synced large binary: %v", err) + } + if !bytes.Equal(received, binary) { + t.Fatal("large binary content differs after blob sync") + } + assertNoUnpushedOps(t, appB) + if err := os.WriteFile(filepath.Join(rootB, "Shared", "external.txt"), []byte("external while closed"), 0o644); err != nil { t.Fatalf("offline external update: %v", err) } diff --git a/internal/core/files/service.go b/internal/core/files/service.go index 4f46dcd..cd2127b 100644 --- a/internal/core/files/service.go +++ b/internal/core/files/service.go @@ -4,6 +4,7 @@ import ( "encoding/base64" "encoding/json" "fmt" + "io" "io/fs" "mime" "os" @@ -201,6 +202,85 @@ func (s *Service) WriteVaultFileBytes(relativePath string, dataBase64 string, op return s.writeVaultFileData(relativePath, data, options) } +// WriteVaultFileFromPath streams a verified temporary file into the vault and +// replaces the destination atomically. It is used by core sync so a Blob never +// needs to be base64-decoded or held in memory. +func (s *Service) WriteVaultFileFromPath(relativePath, sourcePath string, options WriteOptions) error { + source, err := os.Open(sourcePath) + if err != nil { + return err + } + defer source.Close() + info, err := source.Stat() + if err != nil { + return err + } + if !info.Mode().IsRegular() { + return fmt.Errorf("blob source is not a regular file") + } + root, rel, full, err := s.resolveFile(relativePath) + if err != nil { + return err + } + if err := rejectSymlinkPath(root, rel, true); err != nil { + return err + } + parent := filepath.Dir(full) + if info, err := os.Stat(parent); err != nil { + if os.IsNotExist(err) { + return fmt.Errorf("parent-not-found: %s", pathDir(rel)) + } + return err + } else if !info.IsDir() { + return fmt.Errorf("parent-not-directory: %s", pathDir(rel)) + } + existing, err := os.Lstat(full) + if err == nil { + if existing.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("symlink-not-allowed: %s", rel) + } + if !existing.Mode().IsRegular() { + return fmt.Errorf("not-regular-file: %s", rel) + } + if !options.Overwrite { + return fmt.Errorf("conflict: %s", rel) + } + } else if os.IsNotExist(err) { + if !options.CreateIfMissing { + return fmt.Errorf("not-found: %s", rel) + } + } else { + return err + } + tmp, err := os.CreateTemp(parent, ".verstak-write-*") + if err != nil { + return err + } + tmpPath := tmp.Name() + cleanup := true + defer func() { + if cleanup { + _ = os.Remove(tmpPath) + } + }() + if _, err := io.Copy(tmp, source); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + if err := os.Rename(tmpPath, full); err != nil { + return err + } + cleanup = false + return nil +} + func (s *Service) writeVaultFileData(relativePath string, data []byte, options WriteOptions) error { root, rel, full, err := s.resolveFile(relativePath) if err != nil { diff --git a/internal/core/sync/client.go b/internal/core/sync/client.go index 9d10086..93b062d 100644 --- a/internal/core/sync/client.go +++ b/internal/core/sync/client.go @@ -2,6 +2,8 @@ package sync import ( "bytes" + "crypto/sha256" + "encoding/hex" "encoding/json" "fmt" "io" @@ -216,68 +218,147 @@ func (c *Client) Push(ops []Op) (*PushResponse, error) { // PullRequest is the payload for POST /sync/pull. type PullRequest struct { SinceSequence int `json:"since_sequence"` + PageLimit int `json:"page_limit,omitempty"` } // PullResponse is the response from POST /sync/pull. type PullResponse struct { - ServerSequence int `json:"server_sequence"` - Ops []Op `json:"ops"` + ServerSequence int `json:"server_sequence"` + PageLastSequence int `json:"page_last_sequence"` + HasMore bool `json:"has_more"` + Ops []Op `json:"ops"` +} + +// BlobReference is the only binary content representation permitted inside a +// sync operation. The bytes travel via the Blob API, never payload_json. +type BlobReference struct { + SHA256 string `json:"sha256"` + Size int64 `json:"size"` +} + +// ServerError carries a public stable error code. UI layers map Code to their +// own localized wording and must not rely on a server diagnostic string. +type ServerError struct { + Status int + Code string +} + +func (e *ServerError) Error() string { + if e.Code == "" { + return fmt.Sprintf("sync-server:request_failed (HTTP %d)", e.Status) + } + return fmt.Sprintf("sync-server:%s (HTTP %d)", e.Code, e.Status) } // Pull fetches remote operations since a given sequence. func (c *Client) Pull(sinceSequence int) (*PullResponse, error) { - req := PullRequest{SinceSequence: sinceSequence} + return c.PullPage(sinceSequence, 0) +} + +// PullPage fetches one bounded ordered page. The caller advances its durable +// cursor only after every returned operation has applied successfully. +func (c *Client) PullPage(sinceSequence, pageLimit int) (*PullResponse, error) { + req := PullRequest{SinceSequence: sinceSequence, PageLimit: pageLimit} var resp PullResponse if err := c.post("/api/v1/sync/pull", req, &resp); err != nil { return nil, err } + // Servers before pull pagination did not include page_last_sequence. Keep + // the desktop compatible during rolling upgrades without ever advancing + // beyond an operation actually present in the response. + if resp.PageLastSequence == 0 && len(resp.Ops) > 0 { + resp.PageLastSequence = resp.Ops[len(resp.Ops)-1].ServerSequence + } return &resp, nil } -// UploadBlob uploads a file to the server and returns its SHA-256. -func (c *Client) UploadBlob(localPath string) (sha256 string, err error) { - var b bytes.Buffer - w := multipart.NewWriter(&b) - fw, err := w.CreateFormFile("file", filepath.Base(localPath)) +// UploadBlob streams a local file through a multipart pipe. It keeps the +// process memory bounded even when the file is many times larger than an +// inline sync payload. +func (c *Client) UploadBlob(localPath string) (BlobReference, error) { + info, err := os.Stat(localPath) if err != nil { - return "", err + return BlobReference{}, err } - f, err := os.Open(localPath) - if err != nil { - return "", err + if !info.Mode().IsRegular() { + return BlobReference{}, fmt.Errorf("blob source is not a regular file") } - defer f.Close() - if _, err := io.Copy(fw, f); err != nil { - return "", err - } - w.Close() + reader, writer := io.Pipe() + multipartWriter := multipart.NewWriter(writer) + writeDone := make(chan error, 1) + go func() { + defer func() { + _ = writer.Close() + }() + part, err := multipartWriter.CreateFormFile("file", filepath.Base(localPath)) + if err == nil { + file, openErr := os.Open(localPath) + if openErr != nil { + err = openErr + } else { + _, err = io.Copy(part, file) + closeErr := file.Close() + if err == nil { + err = closeErr + } + } + } + if closeErr := multipartWriter.Close(); err == nil { + err = closeErr + } + if err != nil { + _ = writer.CloseWithError(err) + } + writeDone <- err + }() - req, err := http.NewRequest("POST", c.ServerURL+"/api/v1/blobs/", &b) + req, err := http.NewRequest("POST", c.ServerURL+"/api/v1/blobs/", reader) if err != nil { - return "", err + _ = reader.Close() + return BlobReference{}, err } - req.Header.Set("Content-Type", w.FormDataContentType()) + req.Header.Set("Content-Type", multipartWriter.FormDataContentType()) req.Header.Set("Authorization", "Bearer "+c.bearerToken()) resp, err := c.HTTP.Do(req) if err != nil { - return "", err + _ = reader.Close() + <-writeDone + return BlobReference{}, err } defer resp.Body.Close() - - var result struct { - SHA256 string `json:"sha256"` - Size int `json:"size"` + if resp.StatusCode >= http.StatusBadRequest { + writeErr := <-writeDone + if writeErr != nil { + return BlobReference{}, writeErr + } + return BlobReference{}, c.readErrorBody(resp, resp.StatusCode) } + var result BlobReference if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return "", err + return BlobReference{}, err } - return result.SHA256, nil + if err := <-writeDone; err != nil { + return BlobReference{}, err + } + if result.Size != info.Size() || !validBlobSHA256(result.SHA256) { + return BlobReference{}, fmt.Errorf("invalid blob upload response") + } + return result, nil } // DownloadBlob downloads a blob by SHA-256 hash. -func (c *Client) DownloadBlob(sha256, destPath string) error { - req, err := http.NewRequest("GET", c.ServerURL+"/api/v1/blobs/"+sha256, nil) +func (c *Client) DownloadBlob(shaHex, destPath string) error { + return c.DownloadBlobVerified(shaHex, -1, destPath) +} + +// DownloadBlobVerified streams to a temporary file, verifies the announced +// hash and size, and only then atomically makes the file visible to the vault. +func (c *Client) DownloadBlobVerified(shaHex string, expectedSize int64, destPath string) error { + if !validBlobSHA256(shaHex) || expectedSize < -1 { + return fmt.Errorf("invalid blob reference") + } + req, err := http.NewRequest("GET", c.ServerURL+"/api/v1/blobs/"+shaHex, nil) if err != nil { return err } @@ -289,17 +370,65 @@ func (c *Client) DownloadBlob(sha256, destPath string) error { } defer resp.Body.Close() - if resp.StatusCode != 200 { - return fmt.Errorf("download blob: HTTP %d", resp.StatusCode) + if resp.StatusCode != http.StatusOK { + return c.readErrorBody(resp, resp.StatusCode) } - out, err := os.Create(destPath) + if expectedSize >= 0 && resp.ContentLength >= 0 && resp.ContentLength != expectedSize { + return fmt.Errorf("download blob: size mismatch") + } + if err := os.MkdirAll(filepath.Dir(destPath), 0o750); err != nil { + return err + } + out, err := os.CreateTemp(filepath.Dir(destPath), ".verstak-blob-*") if err != nil { return err } - defer out.Close() - _, err = io.Copy(out, resp.Body) - return err + tmpPath := out.Name() + cleanup := true + defer func() { + if cleanup { + _ = os.Remove(tmpPath) + } + }() + hash := sha256.New() + limit := int64(1<<63 - 1) + if expectedSize >= 0 { + limit = expectedSize + 1 + } + written, err := io.Copy(io.MultiWriter(out, hash), io.LimitReader(resp.Body, limit)) + if err != nil { + _ = out.Close() + return err + } + if expectedSize >= 0 && written != expectedSize { + _ = out.Close() + return fmt.Errorf("download blob: size mismatch") + } + if actual := hex.EncodeToString(hash.Sum(nil)); actual != shaHex { + _ = out.Close() + return fmt.Errorf("download blob: SHA-256 mismatch") + } + if err := out.Sync(); err != nil { + _ = out.Close() + return err + } + if err := out.Close(); err != nil { + return err + } + if err := os.Rename(tmpPath, destPath); err != nil { + return err + } + cleanup = false + return nil +} + +func validBlobSHA256(value string) bool { + if len(value) != sha256.Size*2 { + return false + } + _, err := hex.DecodeString(value) + return err == nil } func minInt(a, b int) int { @@ -430,5 +559,11 @@ func (c *Client) readErrorBody(resp *http.Response, statusCode int) error { if strings.Contains(lower, " maxOperationFileBytes { - return nil, fmt.Errorf("file-too-large: %s", entry.Path) - } - if hash, err := sha256File(filepath.Join(vaultRoot, filepath.FromSlash(entry.Path))); err != nil { + path := filepath.Join(vaultRoot, filepath.FromSlash(entry.Path)) + if hash, err := sha256File(path); err != nil { return nil, err } else if hash != entry.Hash { return nil, fmt.Errorf("file changed during scan: %s", entry.Path) } - return filePayload(entry.Path, data, entry.Hash), nil + if entry.Size > corefiles.MaxTextFileBytes { + if err := cacheBlob(path, BlobCachePath(vaultRoot, entry.Hash), entry.Hash); err != nil { + return nil, err + } + payload["blob"] = map[string]interface{}{"sha256": entry.Hash, "size": entry.Size} + return payload, nil + } + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + if isSyncText(data) { + payload["content"] = string(data) + return payload, nil + } + if err := cacheBlob(path, BlobCachePath(vaultRoot, entry.Hash), entry.Hash); err != nil { + return nil, err + } + payload["blob"] = map[string]interface{}{"sha256": entry.Hash, "size": entry.Size} + return payload, nil +} + +func cacheBlob(source, destination, wantHash string) error { + if info, err := os.Lstat(destination); err == nil { + if !info.Mode().IsRegular() { + return fmt.Errorf("blob cache target is not a regular file") + } + if hash, err := sha256File(destination); err == nil && hash == wantHash { + return nil + } + if err := os.Remove(destination); err != nil { + return err + } + } else if !os.IsNotExist(err) { + return err + } + if err := os.MkdirAll(filepath.Dir(destination), 0o750); err != nil { + return err + } + in, err := os.Open(source) + if err != nil { + return err + } + defer in.Close() + tmp, err := os.CreateTemp(filepath.Dir(destination), ".blob-*") + if err != nil { + return err + } + tmpPath := tmp.Name() + cleanup := true + defer func() { + if cleanup { + _ = os.Remove(tmpPath) + } + }() + hash := sha256.New() + if _, err := io.Copy(io.MultiWriter(tmp, hash), in); err != nil { + _ = tmp.Close() + return err + } + if actual := hex.EncodeToString(hash.Sum(nil)); actual != wantHash { + _ = tmp.Close() + return fmt.Errorf("file changed during blob staging") + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + if err := os.Rename(tmpPath, destination); err != nil { + return err + } + cleanup = false + return nil } func newSnapshotOp(deviceID, entityType, entityID, opType string, payload interface{}) Op { @@ -763,19 +841,6 @@ func pathDepth(path string) int { return strings.Count(path, "/") + 1 } -// filePayload encodes an already bounded file without changing its content -// representation. UTF-8 files use plain text up to the existing text limit; -// other supported files use the current bounded base64 transport. -func filePayload(path string, data []byte, hash string) map[string]string { - payload := map[string]string{"path": path, "contentHash": hash} - if int64(len(data)) <= corefiles.MaxTextFileBytes && isSyncText(data) { - payload["content"] = string(data) - return payload - } - payload["dataBase64"] = base64.StdEncoding.EncodeToString(data) - return payload -} - func isSyncText(data []byte) bool { if !utf8.Valid(data) { return false diff --git a/internal/core/sync/snapshot_test.go b/internal/core/sync/snapshot_test.go index 5cd16b0..eabe108 100644 --- a/internal/core/sync/snapshot_test.go +++ b/internal/core/sync/snapshot_test.go @@ -9,6 +9,7 @@ import ( "testing" "github.com/google/uuid" + corefiles "github.com/verstak/verstak-desktop/internal/core/files" ) func TestScanAndRecordTracksExternalWorkspaceLifecycleByIdentity(t *testing.T) { @@ -191,6 +192,41 @@ func TestScanAndRecordBaselinesThenRecordsExternalChanges(t *testing.T) { } } +func TestSnapshotUsesBlobReferenceForBinaryFileBeyondInlineLimit(t *testing.T) { + root := t.TempDir() + service := NewService(root, "device-a") + if _, err := service.ScanAndRecord(); err != nil { + t.Fatalf("baseline: %v", err) + } + data := make([]byte, corefiles.MaxBinaryReadBytes+1) + for i := range data { + data[i] = byte(i % 251) + } + if err := os.WriteFile(filepath.Join(root, "large.bin"), data, 0o600); err != nil { + t.Fatal(err) + } + if _, err := service.ScanAndRecord(); err != nil { + t.Fatalf("scan binary: %v", err) + } + ops := unpushedOps(t, service) + if len(ops) != 1 { + t.Fatalf("operations = %#v, want one file create", ops) + } + var payload struct { + DataBase64 *string `json:"dataBase64"` + Blob *struct { + SHA256 string `json:"sha256"` + Size int64 `json:"size"` + } `json:"blob"` + } + if err := json.Unmarshal([]byte(ops[0].PayloadJSON), &payload); err != nil { + t.Fatal(err) + } + if payload.DataBase64 != nil || payload.Blob == nil || payload.Blob.Size != int64(len(data)) || payload.Blob.SHA256 == "" { + t.Fatalf("binary sync payload = %s, want blob reference without base64", ops[0].PayloadJSON) + } +} + func TestScanAndRecordNeverTreatsInitialFilesAsDeletes(t *testing.T) { root := t.TempDir() if err := os.Mkdir(filepath.Join(root, "Existing"), 0o755); err != nil { @@ -362,7 +398,10 @@ func TestScanAndRecordSkipsReservedTemporaryAndSymlinkPaths(t *testing.T) { func TestScanAndRecordKeepsUnsupportedFileUnresolved(t *testing.T) { root := t.TempDir() path := filepath.Join(root, "too-large.bin") - if err := os.WriteFile(path, make([]byte, maxOperationFileBytes+1), 0o600); err != nil { + if err := os.WriteFile(path, nil, 0o600); err != nil { + t.Fatal(err) + } + if err := os.Truncate(path, maxOperationFileBytes+1); err != nil { t.Fatal(err) } service := NewService(root, "device-a") diff --git a/scripts/smoke-real-sync.sh b/scripts/smoke-real-sync.sh index d288618..0dd2a98 100755 --- a/scripts/smoke-real-sync.sh +++ b/scripts/smoke-real-sync.sh @@ -50,10 +50,10 @@ NOW="$(date -u +%Y-%m-%dT%H:%M:%SZ)" sqlite3 "$DATA_DIR/server.db" " INSERT INTO server_users (id, username, email, password_hash, confirmed, created_at) VALUES ('smoke-user', 'smoke-user', 'smoke@example.test', 'unused', 1, '$NOW'); -INSERT INTO server_devices (id, name, api_key, user_id, vault_id, last_seen, created_at) -VALUES ('smoke-device-a', 'Smoke Device A', 'smoke-key-a', 'smoke-user', 'smoke-vault', '$NOW', '$NOW'); -INSERT INTO server_devices (id, name, api_key, user_id, vault_id, last_seen, created_at) -VALUES ('smoke-device-b', 'Smoke Device B', 'smoke-key-b', 'smoke-user', 'smoke-vault', '$NOW', '$NOW'); +INSERT INTO server_devices (id, name, api_key, legacy_api_key, user_id, vault_id, last_seen, created_at) +VALUES ('smoke-device-a', 'Smoke Device A', 'smoke-key-a', 1, 'smoke-user', 'smoke-vault', '$NOW', '$NOW'); +INSERT INTO server_devices (id, name, api_key, legacy_api_key, user_id, vault_id, last_seen, created_at) +VALUES ('smoke-device-b', 'Smoke Device B', 'smoke-key-b', 1, 'smoke-user', 'smoke-vault', '$NOW', '$NOW'); " (