fix(server): 修复 S3 未知长度上传兼容性与分页探测误报

llmdoc/meta.json

@@ -66,7 +66,7 @@
       "validatedRevision": "2b45dc99c607f59351d214845fe8e0c6b912f478"
     },
     "store/storage-backends.mdx": {
-      "validatedRevision": "2b45dc99c607f59351d214845fe8e0c6b912f478"
+      "validatedRevision": "69eeb627035f9dc6501a316638a137bec986a8d5"
     }
   },
   "convergence": {

llmdoc/store/storage-backends.mdx

@@ -1,5 +1,5 @@
 ---
-description: S3 后端管理：不可变 backendId、active 默认指针、Context 固定绑定、credential generation、真实能力探测与 socket DNS 安全。
+description: S3 后端管理：不可变 backendId、active 默认指针、Context 固定绑定、credential generation、未知长度上传的临时磁盘与 deadline、真实能力探测及 socket DNS 安全。
 kind: architecture
 relations:
   related:
@@ -29,6 +29,8 @@ code:
     - packages/dashboard/src/pages/system/forms/StorageConnectionFields.tsx
     - packages/dashboard/src/pages/system/forms/storageConnection.ts
     - packages/server/test/s3ObjectStore.integration.test.ts
+    - packages/server/test/s3Objects.test.ts
+    - packages/server/test/s3Probe.test.ts
     - packages/server/test/managedSecurity.test.ts
 ---
 
@@ -61,6 +63,14 @@ Node 使用官方 AWS S3 SDK，配置 path-style、禁自动 redirect、单次
 
 test 在随机隔离 namespace 执行真实 PUT/HEAD/GET/DELETE、metadata、空/流式对象、特殊字符 key、分页 continuation，并以并发 `If-None-Match` / `If-Match` 写证明对象服务的原子条件语义。错误 ETag 必須拒绝，probe cleanup 是激活条件之一。`HEAD → PUT` 不能模拟原子 compare-and-set；HTTP mock 或能创建 bucket 不构成兼容证据。
 
+分页检查使用独立子 namespace 与成功写入的预期集合，不能把其他检查的失败写入尝试当作分页缺失；cleanup 则仍清理所有尝试过的 key，包括写入后才报错的对象。
+
+能力检查失败时，服务端 console 只记录固定检查名与严格白名单归一的原因；任意异常消息、cause、key、endpoint 和凭证均不能回显。返回的布尔检查结果仍决定激活资格，诊断用于区分检查语义失败与传输异常，不替代真实兼容验证。
+
 标准 Node driver 当前不提供 presign，因此 default Store 使用流式 relay，平台 S3 Context 不广告可选 direct-upload。core/neutral Store client 仍定义严格 exact-size direct 契约供显式能力 driver 使用，但未实现的浏览器 CORS/直传不能标记已交付。当前是 single PUT，无 multipart/resumable。
 
+未知长度的上传流先以背压写入系统临时目录，完整接收后按真实 `Content-Length` 执行 single PUT，保留原子条件头。这样避免把整份对象缓存在进程堆上；byte cap 在接收过程中执行，同一个 deadline 覆盖暂存、PUT 与确认 HEAD。临时目录与文件权限分别为 `0700`、`0600`，正常完成或错误退出均清理；进程崩溃后的磁盘回收仍需部署环境承担。
+
+这条路径需要可写且容量足够的临时磁盘，并增加完整接收后才能开始远端上传的等待。临时文件是传输暂存，不是部署级本地 ObjectStore；持久对象仍由绑定的 S3 后端保存。兼容判断继续以当前部署的真实全项探测为准，不能从单项布尔失败推定厂商能力或唯一根因。
+
 API、`tb storage` 与 Dashboard 存储后端页提供同权 list/get/add/test/activate/credential update/delete。Context CLI 的 backend 选择与 Dashboard 字段必须落到相同注册 payload；普通 SK 不因拥有 registry read/write 就获得后端管理权限。对字节资源的真实验证每轮有界，保留脱敏证据；不得把 probe namespace 或签名 URL 写入日志。

packages/server/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@tool-bridge/server",
-  "version": "0.22.0",
+  "version": "0.22.1",
   "description": "Self-hosted Tool Bridge Node server with PostgreSQL, S3 object storage and WebSocket device channels",
   "type": "module",
   "license": "MIT",

packages/server/src/s3Objects.ts

@@ -15,9 +15,14 @@ import {
   type ObjectStore,
   TBError,
 } from '@tool-bridge/core'
+import { createReadStream, createWriteStream } from 'node:fs'
 import { NodeHttpHandler } from '@smithy/node-http-handler'
+import { finished, pipeline } from 'node:stream/promises'
 /** Official AWS SDK protocol adapter, deliberately confined to the Node host. */
 import { Readable, Transform } from 'node:stream'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
 import { s3Network, type S3NetworkOptions } from './s3Network'
 
 export interface S3ObjectStoreOptions extends S3NetworkOptions {
@@ -94,10 +99,11 @@ function bounded(source: Readable, maxBytes: number): Readable {
   return output
 }
 
-function uploadBody(
+async function uploadBody(
   body: ObjectBody,
   maxBytes: number,
-): { body: Buffer | Readable, length?: number } {
+  signal: AbortSignal,
+): Promise<{ body: Buffer | Readable, dispose(): Promise<void>, length: number }> {
   if (
     typeof body === 'string'
     || body instanceof Uint8Array
@@ -110,16 +116,39 @@ function uploadBody(
           ? Buffer.from(body)
           : Buffer.from(body.buffer, body.byteOffset, body.byteLength)
     if (value.length > maxBytes) throw limitError()
-    return { body: value, length: value.length }
+    return { body: value, length: value.length, dispose: async () => {} }
   }
   const reader = body.getReader()
+  let total = 0
+  let complete = false
+  let cancelRequested = false
+  const cancel = () => {
+    if (complete || cancelRequested) return
+    cancelRequested = true
+    void reader.cancel?.(signal.reason).catch(() => {})
+  }
+  async function readNext() {
+    signal.throwIfAborted()
+    let onAbort: () => void = () => {}
+    const aborted = new Promise<never>((_, reject) => {
+      onAbort = () => {
+        cancel()
+        reject(signal.reason)
+      }
+      signal.addEventListener('abort', onAbort, { once: true })
+    })
+    try {
+      return await Promise.race([reader.read(), aborted])
+    } finally {
+      // A shared, never-resolved abort promise would retain a reaction per chunk.
+      signal.removeEventListener('abort', onAbort)
+    }
+  }
   const stream = Readable.from(
     (async function* () {
-      let complete = false
-      let total = 0
       try {
         for (;;) {
-          const { done, value } = await reader.read()
+          const { done, value } = await readNext()
           if (done) {
             complete = true
             break
@@ -130,17 +159,37 @@ function uploadBody(
           yield value
         }
       } finally {
-        if (!complete) await reader.cancel?.().catch(() => {})
-        reader.releaseLock()
+        cancel()
       }
     })(),
     { objectMode: false, highWaterMark: 64 * 1024 },
   )
-  // Ensure cancellation reaches a reader whose read() is still pending.
-  stream.on('close', () => {
-    void reader.cancel?.().catch(() => {})
-  })
-  return { body: stream }
+  let directory: string | undefined
+  try {
+    // Use an exact Content-Length instead of relying on unknown-length chunked
+    // PUT support. Spool with backpressure rather than buffering maxBytes in RAM.
+    directory = await mkdtemp(join(tmpdir(), 'tb-s3-upload-'))
+    const path = join(directory, 'body')
+    await pipeline(stream, createWriteStream(path, { flags: 'wx', mode: 0o600 }), { signal })
+    const file = createReadStream(path)
+    const uploadDirectory = directory
+    return {
+      body: file,
+      length: total,
+      async dispose() {
+        file.destroy()
+        await finished(file, { cleanup: true }).catch(() => {})
+        await rm(uploadDirectory, { recursive: true, force: true })
+      },
+    }
+  } catch (error) {
+    stream.destroy()
+    if (directory) await rm(directory, { recursive: true, force: true })
+    throw error
+  } finally {
+    cancel()
+    reader.releaseLock()
+  }
 }
 
 export function createS3ObjectStore(
@@ -237,20 +286,21 @@ export function createS3ObjectStore(
     }
   }
 
-  const head = async (key: string): Promise<ObjectMeta | null> => {
+  const head = async (key: string, signal?: AbortSignal): Promise<ObjectMeta | null> => {
     try {
-      return await run('HEAD', async signal =>
+      const execute = async (signal: AbortSignal) =>
         meta(
           key,
           await client.send(
             new HeadObjectCommand({ Bucket: config.bucket, Key: key }),
             { abortSignal: signal },
           ),
-        ),
-      )
+        )
+      return signal ? await execute(signal) : await run('HEAD', execute)
     } catch (error) {
-      if (isTBError(error) && error.code === 'not_found') return null
-      throw error
+      const normalized = s3Error('HEAD', error)
+      if (normalized.code === 'not_found') return null
+      throw normalized
     }
   }
 
@@ -298,51 +348,45 @@ export function createS3ObjectStore(
           'invalid_argument',
           'S3 conditional write modes are mutually exclusive',
         )
-      const upload = uploadBody(body, maxBytes)
-      let bodyError: Error | undefined
-      const uploadController = new AbortController()
-      if (upload.body instanceof Readable)
-        upload.body.on('error', (error) => {
-          bodyError = error
-          uploadController.abort()
-        })
-      try {
-        await run('PUT', signal =>
-          client.send(
-            new PutObjectCommand({
-              Bucket: config.bucket,
-              Key: key,
-              Body: upload.body,
-              ContentLength: upload.length,
-              ContentType: opts?.contentType,
-              Metadata: opts?.metadata,
-              IfNoneMatch: opts?.ifNoneMatch,
-              IfMatch:
+      return run('PUT', async (signal) => {
+        const upload = await uploadBody(body, maxBytes, signal)
+        const uploadController = new AbortController()
+        if (upload.body instanceof Readable)
+          upload.body.on('error', () => uploadController.abort())
+        try {
+          try {
+            await client.send(
+              new PutObjectCommand({
+                Bucket: config.bucket,
+                Key: key,
+                Body: upload.body,
+                ContentLength: upload.length,
+                ContentType: opts?.contentType,
+                Metadata: opts?.metadata,
+                IfNoneMatch: opts?.ifNoneMatch,
+                IfMatch:
                 opts?.ifMatchEtag !== undefined
                   ? `"${opts.ifMatchEtag}"`
                   : undefined,
-            }),
-            { abortSignal: AbortSignal.any([signal, uploadController.signal]) },
-          ),
-        )
-      } catch (error) {
-        if (bodyError && isTBError(bodyError)) throw bodyError
-        if (
-          opts?.ifMatchEtag !== undefined
-          && isTBError(error)
-          && error.code === 'not_found'
-        )
-          throw new TBError('conflict', 'S3 PUT condition failed')
-        throw error
-      } finally {
-        if (upload.body instanceof Readable) upload.body.destroy()
-      }
-      const stored = await head(key)
-      if (!stored)
-        throw new TBError('unavailable', 'S3 PUT was not observable by HEAD', {
-          retryable: true,
-        })
-      return stored
+              }),
+              { abortSignal: AbortSignal.any([signal, uploadController.signal]) },
+            )
+          } catch (error) {
+            const normalized = s3Error('PUT', error)
+            if (opts?.ifMatchEtag !== undefined && normalized.code === 'not_found')
+              throw new TBError('conflict', 'S3 PUT condition failed')
+            throw normalized
+          }
+          const stored = await head(key, signal)
+          if (!stored)
+            throw new TBError('unavailable', 'S3 PUT was not observable by HEAD', {
+              retryable: true,
+            })
+          return stored
+        } finally {
+          await upload.dispose()
+        }
+      })
     },
     async delete(key) {
       try {

packages/server/src/s3Probe.ts

@@ -8,6 +8,15 @@ export interface S3ProbeResult {
   cleanupSucceeded: boolean
 }
 
+function diagnosticReason(error: unknown): string {
+  if (!isTBError(error)) return 'unexpected_error'
+  // TBError can also originate from an upload source. Only known adapter messages
+  // are safe; never log arbitrary messages, causes, keys, endpoints or credentials.
+  const normalized = /^(?:S3 (?:PUT|HEAD|GET|LIST|DELETE) (?:failed(?: \([1-5]\d{2}\))?|was denied|object not found|condition failed)|S3 GET deadline exceeded|S3 PUT was not observable by HEAD)$/.exec(error.message)
+  // JS $ also matches before a final newline, so require a complete match.
+  return normalized?.[0] === error.message ? error.message : 'operation_failed'
+}
+
 /** Destructive only inside a fresh, unguessable probe namespace in the selected bucket. */
 export async function probeS3ObjectStore(
   config: S3StoreConfig,
@@ -40,11 +49,15 @@ export async function probeS3ObjectStore(
     name: string,
     run: () => Promise<boolean>,
   ): Promise<void> {
+    let reason = 'unexpected_result'
     try {
       checks[name] = await run()
-    } catch {
+    } catch (error) {
       checks[name] = false
+      reason = diagnosticReason(error)
     }
+    if (!checks[name])
+      console.warn('S3 capability probe check failed', { check: name, reason })
   }
   async function read(name: string): Promise<string | undefined> {
     const object = await store.get(name)
@@ -161,11 +174,20 @@ export async function probeS3ObjectStore(
       return true
     })
     await check('pagination', async () => {
+      // Other checks may fail before or after their PUT reaches the backend.
+      // Keep listing expectations independent from the cleanup attempt ledger.
+      const paginationPrefix = `${prefix}pagination/`
+      const expected = new Set<string>()
+      for (let index = 0; index < 5; index++) {
+        const target = key(`pagination/item-${index}`)
+        await store.put(target, `page-${index}`, { ifNoneMatch: '*' })
+        expected.add(target)
+      }
       const found = new Set<string>()
       const cursors = new Set<string>()
       let cursor: string | undefined
       do {
-        const page = await store.list(prefix, { cursor, limit: 2 })
+        const page = await store.list(paginationPrefix, { cursor, limit: 2 })
         for (const item of page.items) {
           if (!('key' in item) || found.has(item.key)) return false
           found.add(item.key)
@@ -177,8 +199,8 @@ export async function probeS3ObjectStore(
         }
       } while (cursor)
       return (
-        found.size === keys.size
-        && [...keys].every(item => found.has(item))
+        found.size === expected.size
+        && [...expected].every(item => found.has(item))
         && cursors.size > 0
       )
     })