Release v2.48.0: Cloud conflict decisions save one note and stay saved, queued conflicts merge when they can, interrupted uploads clean up, Escape keeps Quick Capture open

apps/desktop/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@zennotes/desktop",
   "productName": "ZenNotes",
-  "version": "2.47.0",
+  "version": "2.48.0",
   "description": "ZenNotes desktop shell",
   "private": true,
   "main": "./out/main/index.js",

apps/desktop/src/main/cloud-sync-client.test.ts

@@ -1,15 +1,19 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
+import * as nodeFs from 'node:fs'
 import { mkdtemp, rm, writeFile } from 'node:fs/promises'
 import os from 'node:os'
 import path from 'node:path'
 import type { CloudSyncUpsertMutation } from '@zennotes/bridge-contract/cloud-sync'
 import { CloudServiceRequestError, createCloudSyncClient } from './cloud-sync-client'
 import { rememberCloudSyncUploadSource } from './cloud-sync-upload-source'
 
+vi.mock('node:fs', { spy: true })
+
 const INLINE_UPLOAD_LIMIT_BYTES = 5 * 1024 * 1024
 const temporaryDirectories: string[] = []
 
 afterEach(async () => {
+  vi.restoreAllMocks()
   await Promise.all(
     temporaryDirectories
       .splice(0)
@@ -18,6 +22,51 @@ afterEach(async () => {
 })
 
 describe('createCloudSyncClient', () => {
+  it.each(['reject', 'early response'] as const)(
+    'closes the file source before releasing the reservation after an upload %s',
+    async (failure) => {
+      const directory = await mkdtemp(path.join(os.tmpdir(), 'zennotes-upload-failure-'))
+      temporaryDirectories.push(directory)
+      const sourcePath = path.join(directory, 'image.jpg')
+      const bytes = Buffer.alloc(8_100_000, 7)
+      await writeFile(sourcePath, bytes)
+      const mutation = upsertMutation(bytes.length, '')
+      rememberCloudSyncUploadSource(mutation.content, sourcePath)
+      const openFile = vi.spyOn(nodeFs, 'createReadStream')
+      let destroyedBeforeAbort = false
+      const fetchImplementation = vi.fn<typeof fetch>(async (input, options) => {
+        const url = String(input)
+        if (url.endsWith('/uploads') && options?.method === 'POST') {
+          return jsonResponse({
+            data: {
+              id: 'interrupted',
+              operation_id: mutation.operation_id,
+              expected_bytes: bytes.length,
+              upload: { method: 'PUT', url: 'https://objects.example.test/image', headers: {} }
+            }
+          })
+        }
+        if (url === 'https://objects.example.test/image') {
+          if (failure === 'reject') throw new TypeError('fetch failed')
+          return new Response(null, { status: 503 })
+        }
+        if (options?.method === 'DELETE') {
+          destroyedBeforeAbort = openFile.mock.results[0]?.value.destroyed === true
+          return new Response(null, { status: 204 })
+        }
+        throw new Error(`Unexpected request: ${url}`)
+      })
+      try {
+        const client = createCloudSyncClient('https://zennotes.test', 'token', fetchImplementation)
+        await expect(client.mutate('vault', { mutations: [mutation] })).rejects.toThrow()
+        expect(destroyedBeforeAbort).toBe(true)
+      } finally {
+        // Keep the regression itself from leaking the old implementation's reader.
+        for (const result of openFile.mock.results) result.value?.destroy()
+      }
+    }
+  )
+
   it('authenticates requests without exposing the token in the URL', async () => {
     const fetchImplementation = vi.fn<typeof fetch>().mockResolvedValue(
       new Response(JSON.stringify({ data: [] }), {
@@ -212,7 +261,9 @@ describe('createCloudSyncClient', () => {
         )
       }
       if (url.startsWith('https://objects.example.test/')) {
-        uploadedBodies.push(Buffer.from(await new Response(options?.body).arrayBuffer()))
+        const chunks: Uint8Array[] = []
+        for await (const chunk of options?.body as AsyncIterable<Uint8Array>) chunks.push(chunk)
+        uploadedBodies.push(Buffer.concat(chunks))
         return new Response(null, { status: 200 })
       }
       if (url.endsWith('/complete')) {

apps/desktop/src/main/cloud-sync-client.ts

@@ -3,9 +3,8 @@ import {
   type CloudSyncHttpRequest,
   type CloudSyncHttpTransport
 } from '@zennotes/shared-domain/cloud-sync-api'
-import { createReadStream } from 'node:fs'
+import { createReadStream, type ReadStream } from 'node:fs'
 import { stat } from 'node:fs/promises'
-import { Readable } from 'node:stream'
 import type {
   CloudSyncCapacityConflict,
   CloudSyncConflict,
@@ -124,18 +123,25 @@ class DesktopCloudSyncApiClient extends CloudSyncApiClient {
     let response: Response
 
     try {
-      response = await this.fetchImplementation(uploadUrl, {
-        method: upload.method,
-        headers: upload.headers,
-        body: uploadBody.createBody(),
-        signal: AbortSignal.timeout(DIRECT_UPLOAD_TIMEOUT_MS),
-        redirect: 'error',
-        ...(uploadBody.stream ? { duplex: 'half' } : {})
-      })
+      try {
+        response = await this.fetchImplementation(uploadUrl, {
+          method: upload.method,
+          headers: upload.headers,
+          body: uploadBody.createBody(),
+          signal: AbortSignal.timeout(DIRECT_UPLOAD_TIMEOUT_MS),
+          redirect: 'error',
+          ...(uploadBody.stream ? { duplex: 'half' } : {})
+        })
+      } finally {
+        // A server can reject the PUT before reading the file. Stop its reader
+        // before waiting for reservation cleanup (including on timeout/abort).
+        uploadBody.dispose()
+      }
     } catch (error) {
       await this.abortQuietly(vaultId, instruction.id)
       throw error
     }
+    await response.body?.cancel().catch(() => {})
 
     if (!response.ok) {
       await this.abortQuietly(vaultId, instruction.id)
@@ -296,22 +302,29 @@ function uploadRequest(mutation: CloudSyncUpsertMutation): CloudSyncUploadReques
 
 async function prepareDirectUploadBody(
   mutation: CloudSyncUpsertMutation
-): Promise<{ createBody(): FetchBody; stream: boolean }> {
+): Promise<{ createBody(): FetchBody; dispose(): void; stream: boolean }> {
   const sourcePath = cloudSyncUploadSource(mutation.content)
   if (sourcePath) {
     const sourceStats = await stat(sourcePath)
     if (!sourceStats.isFile() || sourceStats.size !== mutation.content.byte_length) {
       throw directUploadSizeMismatch()
     }
+    let reader: ReadStream | undefined
     return {
-      createBody: () => Readable.toWeb(createReadStream(sourcePath)) as unknown as FetchBody,
+      // Node fetch accepts async iterables directly. Avoid the event-based
+      // toWeb adapter: late file events after cancellation can enqueue into
+      // a closed WebStream controller outside the fetch promise's catch.
+      createBody: () => (reader = createReadStream(sourcePath)) as unknown as FetchBody,
+      dispose: () => {
+        reader?.destroy()
+      },
       stream: true
     }
   }
 
   const bytes = uploadBytes(mutation)
   if (bytes.byteLength !== mutation.content.byte_length) throw directUploadSizeMismatch()
-  return { createBody: () => bytes as FetchBody, stream: false }
+  return { createBody: () => bytes as FetchBody, dispose: () => {}, stream: false }
 }
 
 function uploadBytes(mutation: CloudSyncUpsertMutation): Uint8Array {

apps/desktop/src/main/cloud-sync-upload-network.test.ts

@@ -0,0 +1,226 @@
+import { createHash } from 'node:crypto'
+import { once } from 'node:events'
+import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { createServer, type Server } from 'node:http'
+import { tmpdir } from 'node:os'
+import path from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import type { CloudSyncUpsertMutation } from '@zennotes/bridge-contract/cloud-sync'
+import { createCloudSyncClient } from './cloud-sync-client'
+import { rememberCloudSyncUploadSource } from './cloud-sync-upload-source'
+import { createDesktopCloudSyncCoordinator } from './cloud-sync-filesystem'
+
+const servers: Server[] = []
+const directories: string[] = []
+
+afterEach(async () => {
+  await Promise.all(
+    servers.splice(0).map(async (server) => {
+      server.closeAllConnections()
+      await new Promise<void>((resolve) => server.close(() => resolve()))
+    })
+  )
+  await Promise.all(
+    directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true }))
+  )
+})
+
+describe('disk-backed Cloud uploads over HTTP', () => {
+  it('rebuilds an interrupted upload from disk after recreating the sync coordinator', async () => {
+    const fixture = await setup(8_100_000, 'disconnect')
+    const coordinator = () =>
+      createDesktopCloudSyncCoordinator({
+        root: fixture.localRoot,
+        stateDirectory: fixture.stateDirectory,
+        vaultId: 'vault',
+        remote: fixture.client()
+      })
+    await expect(coordinator().sync()).rejects.toThrow()
+    fixture.recover()
+    // Recreate the repository, state store and client: no prior content object
+    // or WeakMap upload source is reused, just the on-disk vault and sync state.
+    expect((await coordinator().sync()).pushed).toBe(1)
+    expect((await coordinator().sync()).pushed).toBe(0)
+    expect(fixture.completions()).toBe(1)
+    expect(sha256(await readFile(path.join(fixture.localRoot, 'assets/image.jpg')))).toBe(
+      fixture.mutation.content.sha256
+    )
+    const downloaded = Buffer.from(await (await fetch(`${fixture.url}/object`)).arrayBuffer())
+    expect(sha256(downloaded)).toBe(fixture.mutation.content.sha256)
+  })
+
+  it.each([8_100_000, 10_000_000])(
+    'round-trips all %i bytes through a real upload and download',
+    async (size) => {
+      const fixture = await setup(size)
+      const result = await fixture.client().mutate('vault', { mutations: [fixture.mutation] })
+      expect(result.acknowledged).toHaveLength(1)
+      const downloaded = Buffer.from(await (await fetch(`${fixture.url}/object`)).arrayBuffer())
+      expect(downloaded.length).toBe(size)
+      expect(sha256(downloaded)).toBe(fixture.mutation.content.sha256)
+    }
+  )
+
+  it.each(['reject', 'disconnect', 'timeout'] as const)(
+    'survives an upload %s and retries the same file with a fresh client',
+    async (failure) => {
+      const fixture = await setup(8_100_000, failure)
+      await expect(
+        fixture.client().mutate('vault', { mutations: [fixture.mutation] })
+      ).rejects.toThrow()
+      expect(fixture.aborts()).toBe(1)
+      expect(fixture.completions()).toBe(0)
+      fixture.recover()
+      // A new client has no in-memory upload state, as after an app restart.
+      const result = await fixture.client().mutate('vault', { mutations: [fixture.mutation] })
+      expect(result.acknowledged).toHaveLength(1)
+      const downloaded = Buffer.from(await (await fetch(`${fixture.url}/object`)).arrayBuffer())
+      expect(sha256(downloaded)).toBe(fixture.mutation.content.sha256)
+    }
+  )
+})
+
+async function setup(
+  size: number,
+  initialFailure: 'reject' | 'disconnect' | 'timeout' | null = null
+) {
+  const directory = await mkdtemp(path.join(tmpdir(), 'zennotes-upload-network-'))
+  directories.push(directory)
+  const localRoot = path.join(directory, 'vault')
+  await mkdir(path.join(localRoot, 'assets'), { recursive: true })
+  const source = path.join(localRoot, 'assets/image.jpg')
+  const bytes = Buffer.alloc(size)
+  for (let index = 0; index < bytes.length; index++) bytes[index] = index % 251
+  await writeFile(source, bytes)
+  const mutation: CloudSyncUpsertMutation = {
+    type: 'upsert',
+    operation_id: 'upload-operation',
+    item_id: 'image',
+    base_revision: 0,
+    path: 'assets/image.jpg',
+    kind: 'binary',
+    content: rememberCloudSyncUploadSource(
+      {
+        encoding: 'base64',
+        data: '',
+        byte_length: bytes.length,
+        sha256: sha256(bytes),
+        media_type: 'image/jpeg'
+      },
+      source
+    )
+  }
+  let failure = initialFailure
+  let stored: Buffer | null = null
+  let aborts = 0
+  let completions = 0
+  let uploadMutation = mutation
+  const server = createServer((request, response) => {
+    request.on('error', () => {})
+    const json = (status: number, body: unknown) => {
+      response.writeHead(status, { 'Content-Type': 'application/json' })
+      response.end(JSON.stringify(body))
+    }
+    if (request.url?.endsWith('/uploads') && request.method === 'POST') {
+      const chunks: Buffer[] = []
+      request.on('data', (chunk: Buffer) => chunks.push(chunk))
+      request.on('end', () => {
+        uploadMutation = JSON.parse(Buffer.concat(chunks).toString()) as CloudSyncUpsertMutation
+        json(201, {
+          data: {
+            id: 'session',
+            operation_id: uploadMutation.operation_id,
+            expected_bytes: bytes.length,
+            upload: {
+              method: 'PUT',
+              url: `${url}/object`,
+              headers: { 'Content-Length': String(bytes.length) }
+            }
+          }
+        })
+      })
+    } else if (request.url === '/object' && request.method === 'PUT') {
+      if (failure === 'reject') {
+        response.writeHead(403, { Connection: 'close' })
+        response.end()
+      } else if (failure === 'disconnect') {
+        request.once('data', () => request.socket.destroy())
+      } else if (failure === 'timeout') {
+        request.pause()
+      } else {
+        const chunks: Buffer[] = []
+        request.on('data', (chunk: Buffer) => chunks.push(chunk))
+        request.on('end', () => {
+          stored = Buffer.concat(chunks)
+          response.writeHead(200)
+          response.end()
+        })
+      }
+    } else if (request.method === 'DELETE') {
+      aborts++
+      response.writeHead(204)
+      response.end()
+    } else if (request.url?.endsWith('/complete')) {
+      completions++
+      request.resume()
+      if (!stored || sha256(stored) !== mutation.content.sha256) {
+        json(422, { error: { message: 'Upload is incomplete' } })
+      } else {
+        json(200, {
+          data: {
+            result: {
+              acknowledged: [
+                {
+                  operation_id: uploadMutation.operation_id,
+                  item_id: uploadMutation.item_id,
+                  revision: 1,
+                  sequence: 1
+                }
+              ],
+              conflicts: [],
+              cursor: 1
+            }
+          }
+        })
+      }
+    } else if (request.url === '/object' && request.method === 'GET' && stored) {
+      response.end(stored)
+    } else if (request.url?.includes('/manifest')) {
+      json(200, { data: [], cursor: 0, next_page: null })
+    } else if (request.url?.includes('/changes')) {
+      json(200, { data: [], cursor: stored ? 1 : 0, has_more: false })
+    } else {
+      json(404, {})
+    }
+  })
+  servers.push(server)
+  server.listen(0, '127.0.0.1')
+  await once(server, 'listening')
+  const address = server.address()
+  if (!address || typeof address === 'string') throw new Error('Missing test server address')
+  const url = `http://127.0.0.1:${address.port}`
+  // Exercise the real fetch cancellation path without waiting the production five-minute timeout.
+  const transport: typeof fetch = (input, options) =>
+    fetch(input, {
+      ...options,
+      ...(failure === 'timeout' && options?.method === 'PUT'
+        ? { signal: AbortSignal.timeout(100) }
+        : {})
+    })
+  return {
+    url,
+    localRoot,
+    stateDirectory: path.join(directory, 'state'),
+    mutation,
+    client: () => createCloudSyncClient(url, 'test-token', transport),
+    recover: () => {
+      failure = null
+    },
+    aborts: () => aborts,
+    completions: () => completions
+  }
+}
+
+function sha256(bytes: Buffer): string {
+  return createHash('sha256').update(bytes).digest('hex')
+}