1- import { test } from 'vitest' ;
1+ import { test , vi } from 'vitest' ;
22import assert from 'node:assert/strict' ;
3+ import crypto from 'node:crypto' ;
4+ import fs from 'node:fs' ;
5+ import { PassThrough , Readable } from 'node:stream' ;
6+ import type { IncomingMessage } from 'node:http' ;
37import { AppError } from '@agent-device/kernel/errors' ;
4- import { finalizeResumableUpload } from '../resumable-upload.ts' ;
8+ import {
9+ beginResumableUpload ,
10+ finalizeResumableUpload ,
11+ receiveResumableUploadChunk ,
12+ } from '../resumable-upload.ts' ;
513
614test ( 'finalizing an unknown upload reports expiry with a recovery hint' , async ( ) => {
715 const error = await finalizeResumableUpload ( 'missing-upload-id' ) . then (
@@ -15,3 +23,98 @@ test('finalizing an unknown upload reports expiry with a recovery hint', async (
1523 assert . equal ( appError . details ?. reason , 'RESOURCE_EXPIRED' ) ;
1624 assert . equal ( typeof appError . details ?. hint , 'string' ) ;
1725} ) ;
26+
27+ test ( 'oversized ranged chunks roll back atomically and can be retried and finalized' , async ( ) => {
28+ const bytes = Buffer . from ( 'ABCDE' ) ;
29+ const uploadId = beginUpload ( bytes ) . uploadId ;
30+ await assert . rejects (
31+ receiveResumableUploadChunk ( {
32+ uploadId,
33+ req : request ( Buffer . from ( 'ABC' ) , { 'content-range' : 'bytes 0-1/5' } ) ,
34+ } ) ,
35+ / p e r m i t t e d b y t e r a n g e / i,
36+ ) ;
37+ assert . deepEqual (
38+ await receiveResumableUploadChunk ( {
39+ uploadId,
40+ req : request ( Buffer . from ( 'AB' ) , { 'content-range' : 'bytes 0-1/5' } ) ,
41+ } ) ,
42+ { complete : false , offset : 2 } ,
43+ ) ;
44+ await receiveResumableUploadChunk ( {
45+ uploadId,
46+ req : request ( Buffer . from ( 'CDE' ) , { 'content-range' : 'bytes 2-4/5' } ) ,
47+ } ) ;
48+ const finalized = await finalizeResumableUpload ( uploadId ) ;
49+ try {
50+ assert . equal ( fs . readFileSync ( finalized . artifactPath , 'utf8' ) , 'ABCDE' ) ;
51+ } finally {
52+ fs . rmSync ( finalized . tempDir , { recursive : true , force : true } ) ;
53+ }
54+ } ) ;
55+
56+ test ( 'an early finalize keeps the upload resumable' , async ( ) => {
57+ const bytes = Buffer . from ( 'resume' ) ;
58+ const uploadId = beginUpload ( bytes ) . uploadId ;
59+ await assert . rejects ( finalizeResumableUpload ( uploadId ) , / i n c o m p l e t e / i) ;
60+ await receiveResumableUploadChunk ( { uploadId, req : request ( bytes ) } ) ;
61+ const finalized = await finalizeResumableUpload ( uploadId ) ;
62+ fs . rmSync ( finalized . tempDir , { recursive : true , force : true } ) ;
63+ } ) ;
64+
65+ test ( 'per-ticket operations serialize while an earlier chunk is paused' , async ( ) => {
66+ const bytes = Buffer . from ( 'ABCD' ) ;
67+ const uploadId = beginUpload ( bytes ) . uploadId ;
68+ const first = requestStream ( { 'content-range' : 'bytes 0-1/4' } ) ;
69+ const second = requestStream ( { 'content-range' : 'bytes 2-3/4' } ) ;
70+ const firstResult = receiveResumableUploadChunk ( { uploadId, req : first } ) ;
71+ const secondResult = receiveResumableUploadChunk ( { uploadId, req : second } ) ;
72+ second . end ( 'CD' ) ;
73+ let secondSettled = false ;
74+ void secondResult . finally ( ( ) => {
75+ secondSettled = true ;
76+ } ) ;
77+ await Promise . resolve ( ) ;
78+ assert . equal ( secondSettled , false ) ;
79+ first . end ( 'AB' ) ;
80+ assert . deepEqual ( await firstResult , { complete : false , offset : 2 } ) ;
81+ assert . deepEqual ( await secondResult , { complete : true , offset : 4 } ) ;
82+ const finalized = await finalizeResumableUpload ( uploadId ) ;
83+ fs . rmSync ( finalized . tempDir , { recursive : true , force : true } ) ;
84+ } ) ;
85+
86+ test ( 'expiry aborts an active receive and invalidates the ticket after rollback' , async ( ) => {
87+ vi . useFakeTimers ( ) ;
88+ try {
89+ const bytes = Buffer . from ( 'AB' ) ;
90+ const uploadId = beginUpload ( bytes ) . uploadId ;
91+ const body = requestStream ( ) ;
92+ const receiving = receiveResumableUploadChunk ( { uploadId, req : body } ) ;
93+ body . write ( 'A' ) ;
94+ await vi . advanceTimersByTimeAsync ( 5 * 60 * 1000 ) ;
95+ await assert . rejects ( receiving , / e x p i r e d / i) ;
96+ await assert . rejects ( finalizeResumableUpload ( uploadId ) , / n o t f o u n d o r e x p i r e d / i) ;
97+ } finally {
98+ vi . useRealTimers ( ) ;
99+ }
100+ } ) ;
101+
102+ function beginUpload ( bytes : Buffer ) : ReturnType < typeof beginResumableUpload > {
103+ return beginResumableUpload ( {
104+ baseUrl : 'http://127.0.0.1:1234' ,
105+ tokenHeaders : { } ,
106+ uploadAttemptId : crypto . randomUUID ( ) ,
107+ sha256 : crypto . createHash ( 'sha256' ) . update ( bytes ) . digest ( 'hex' ) ,
108+ fileName : 'artifact.bin' ,
109+ sizeBytes : bytes . length ,
110+ artifactType : 'file' ,
111+ } ) ;
112+ }
113+
114+ function request ( body : Buffer , headers : Record < string , string > = { } ) : IncomingMessage {
115+ return Object . assign ( Readable . from ( body ) , { headers } ) as IncomingMessage ;
116+ }
117+
118+ function requestStream ( headers : Record < string , string > = { } ) : PassThrough & IncomingMessage {
119+ return Object . assign ( new PassThrough ( ) , { headers } ) as PassThrough & IncomingMessage ;
120+ }
0 commit comments