diff --git a/go_backend/coverage_test_helpers_test.go b/go_backend/coverage_test_helpers_test.go index 8eafb2cd..5cac7bc1 100644 --- a/go_backend/coverage_test_helpers_test.go +++ b/go_backend/coverage_test_helpers_test.go @@ -139,12 +139,18 @@ registerExtension({ return t; }, checkAvailability: function(isrc, name, artist, ids) { - return { available: true, reason: "ok", trackId: "download-track", skipFallback: true }; + return { + available: true, + reason: "ok", + trackId: "download-track", + skipFallback: true, + prepared_context: { token: "prepared", resolvedTrackId: "download-track" } + }; }, getDownloadUrl: function(id, quality) { return { url: "https://example.test/audio.flac", format: "flac", bitDepth: 24, sampleRate: 96000 }; }, - download: function(id, quality, outputPath, onProgress) { + download: function(id, quality, outputPath, onProgress, options) { if (onProgress) onProgress(100); return { success: true, @@ -152,7 +158,9 @@ registerExtension({ alreadyExists: false, bitDepth: 24, sampleRate: 96000, - title: "Downloaded", + title: options && options.preparedContext + ? options.preparedContext.token + : "Downloaded", artist: "Artist", album: "Album", albumArtist: "Album Artist", diff --git a/go_backend/extension_fallback.go b/go_backend/extension_fallback.go index 7022d819..d42b3790 100644 --- a/go_backend/extension_fallback.go +++ b/go_backend/extension_fallback.go @@ -20,6 +20,7 @@ func attemptExtensionDownload( ext *loadedExtension, provider *extensionProviderWrapper, trackID, quality, providerLabel string, + preparedContext map[string]any, applyTitleFallback bool, lastErr *error, lastErrType *string, @@ -49,7 +50,7 @@ func attemptExtensionDownload( SetItemPreparingStage(req.ItemID, "resolving_stream") } - result, err := provider.Download(trackID, quality, outputPath, req.ItemID, func(percent int) { + result, err := provider.DownloadPrepared(trackID, quality, outputPath, req.ItemID, preparedContext, func(percent int) { if req.ItemID != "" { normalized := float64(percent) / 100.0 if normalized < 0 { @@ -250,6 +251,12 @@ func attemptVerifiedResumeBeforeMetadata( trackID, req.Quality, selectedProvider, + func() map[string]any { + if availability == nil { + return nil + } + return availability.PreparedContext + }(), strings.EqualFold(sourceProvider, selectedProvider), &lastErr, &lastErrType, @@ -555,7 +562,11 @@ func DownloadWithExtensionFallback(req DownloadRequest) (*DownloadResponse, erro GoLog("[DownloadWithExtensionFallback] Downloading from source extension with trackID: %s (stopProviderFallback: %v)\n", trackID, stopProviderFallback) - resp, cancelledOuter := attemptExtensionDownload(req, ext, provider, trackID, req.Quality, req.Source, true, &lastErr, &lastErrType, &lastRetryAfterSeconds) + var preparedContext map[string]any + if sourceExtensionAvailability != nil { + preparedContext = sourceExtensionAvailability.PreparedContext + } + resp, cancelledOuter := attemptExtensionDownload(req, ext, provider, trackID, req.Quality, req.Source, preparedContext, true, &lastErr, &lastErrType, &lastRetryAfterSeconds) if cancelledOuter { return nil, ErrDownloadCancelled } @@ -700,7 +711,7 @@ func DownloadWithExtensionFallback(req DownloadRequest) (*DownloadResponse, erro } } - resp, cancelledOuter := attemptExtensionDownload(req, ext, provider, availability.TrackID, fallbackQuality, providerID, false, &lastErr, &lastErrType, &lastRetryAfterSeconds) + resp, cancelledOuter := attemptExtensionDownload(req, ext, provider, availability.TrackID, fallbackQuality, providerID, availability.PreparedContext, false, &lastErr, &lastErrType, &lastRetryAfterSeconds) if cancelledOuter { return nil, ErrDownloadCancelled } diff --git a/go_backend/extension_goja_convert.go b/go_backend/extension_goja_convert.go index b0e8340c..9b323730 100644 --- a/go_backend/extension_goja_convert.go +++ b/go_backend/extension_goja_convert.go @@ -455,6 +455,11 @@ func parseExtensionAvailabilityValue(vm *goja.Runtime, value goja.Value) ExtAvai Reason: gojaObjectString(obj, "reason"), TrackID: gojaObjectString(obj, "track_id", "trackId"), SkipFallback: gojaObjectBool(obj, "skip_fallback", "skipFallback"), + PreparedContext: gojaObjectInterfaceMap( + obj, + "prepared_context", + "preparedContext", + ), } } diff --git a/go_backend/extension_manager.go b/go_backend/extension_manager.go index d407dcca..64226eba 100644 --- a/go_backend/extension_manager.go +++ b/go_backend/extension_manager.go @@ -251,11 +251,12 @@ func (m *extensionManager) loadExtensionFromFileLocked(filePath string) (*loaded } var supportedRuntimeFeatures = map[string]int{ - "signedSession": 3, - "sessionRefresh": 1, - "sessionGrant": 1, - "globalAction": 1, - "webviewAuth": 1, + "signedSession": 3, + "sessionRefresh": 1, + "sessionGrant": 1, + "globalAction": 1, + "webviewAuth": 1, + "downloadSegments": 1, } // validateManifestGates enforces minAppVersion and requiredRuntimeFeatures diff --git a/go_backend/extension_manifest.go b/go_backend/extension_manifest.go index e73a36fe..ebdc95f4 100644 --- a/go_backend/extension_manifest.go +++ b/go_backend/extension_manifest.go @@ -293,6 +293,9 @@ func (m *ExtensionManifest) Validate() error { if m.HasCapability("rawFfmpeg") && !m.Permissions.File { return &ManifestValidationError{Field: "permissions.file", Message: "rawFfmpeg capability requires file permission"} } + if err := validateDownloadTransferCapability(m.Capabilities); err != nil { + return &ManifestValidationError{Field: "capabilities.downloadTransfer", Message: err.Error()} + } return nil } diff --git a/go_backend/extension_provider_supplement_test.go b/go_backend/extension_provider_supplement_test.go index 9e3137f0..fe7e1a67 100644 --- a/go_backend/extension_provider_supplement_test.go +++ b/go_backend/extension_provider_supplement_test.go @@ -64,6 +64,9 @@ func TestExtensionProviderWrapperFullSurface(t *testing.T) { if !availability.Available || availability.TrackID != "download-track" || !availability.SkipFallback { t.Fatalf("availability = %#v", availability) } + if availability.PreparedContext["token"] != "prepared" { + t.Fatalf("prepared context = %#v", availability.PreparedContext) + } progress := []int{} download, err := provider.Download("track-1", "LOSSLESS", filepath.Join(t.TempDir(), "song.flac"), "", func(percent int) { @@ -75,6 +78,17 @@ func TestExtensionProviderWrapperFullSurface(t *testing.T) { if !download.Success || download.Decryption == nil || download.DecryptionKey != "001122" || download.Comment != "https://example.test/album/1" || !download.Explicit || download.AlbumType != "compilation" || download.UPC != "0012345678901" || len(progress) != 1 || progress[0] != 100 { t.Fatalf("download = %#v progress=%v", download, progress) } + preparedDownload, err := provider.DownloadPrepared( + "track-1", + "LOSSLESS", + filepath.Join(t.TempDir(), "prepared.flac"), + "", + availability.PreparedContext, + nil, + ) + if err != nil || preparedDownload == nil || preparedDownload.Title != "prepared" { + t.Fatalf("prepared download = %#v, err=%v", preparedDownload, err) + } lyrics, err := provider.FetchLyrics("Song", "Artist", "Album", 180) if err != nil { diff --git a/go_backend/extension_provider_types.go b/go_backend/extension_provider_types.go index e77ca193..31e1d513 100644 --- a/go_backend/extension_provider_types.go +++ b/go_backend/extension_provider_types.go @@ -87,10 +87,11 @@ type ExtSearchResult struct { } type ExtAvailabilityResult struct { - Available bool `json:"available"` - Reason string `json:"reason,omitempty"` - TrackID string `json:"track_id,omitempty"` - SkipFallback bool `json:"skip_fallback,omitempty"` + Available bool `json:"available"` + Reason string `json:"reason,omitempty"` + TrackID string `json:"track_id,omitempty"` + SkipFallback bool `json:"skip_fallback,omitempty"` + PreparedContext map[string]any `json:"prepared_context,omitempty"` } type DownloadDecryptionInfo struct { diff --git a/go_backend/extension_provider_wrapper.go b/go_backend/extension_provider_wrapper.go index 75d9219b..ff2e8d03 100644 --- a/go_backend/extension_provider_wrapper.go +++ b/go_backend/extension_provider_wrapper.go @@ -565,6 +565,25 @@ const ExtDownloadTimeout = DownloadTimeout // an isolated VM/runtime (not p.vm/p.extension.VMMu) with a progress // callback, which the helper's lock+perf model doesn't cover. func (p *extensionProviderWrapper) Download(trackID, quality, outputPath, itemID string, onProgress func(percent int)) (*ExtDownloadResult, error) { + return p.DownloadPrepared( + trackID, + quality, + outputPath, + itemID, + nil, + onProgress, + ) +} + +// DownloadPrepared passes the opaque context returned by checkAvailability to +// the isolated download runtime. Existing extensions remain compatible because +// JavaScript ignores the additional options argument; extensions that opt in +// can reuse already-resolved metadata or stream preparation. +func (p *extensionProviderWrapper) DownloadPrepared( + trackID, quality, outputPath, itemID string, + preparedContext map[string]any, + onProgress func(percent int), +) (*ExtDownloadResult, error) { if !p.extension.Manifest.IsDownloadProvider() { return nil, fmt.Errorf("extension '%s' is not a download provider", p.extension.ID) } @@ -622,8 +641,20 @@ func (p *extensionProviderWrapper) Download(trackID, quality, outputPath, itemID } jsStartedAt := time.Now() + downloadOptions := map[string]any{} + if len(preparedContext) > 0 { + downloadOptions["preparedContext"] = preparedContext + } result, err := runGojaCallWithTimeoutAndRecover(vm, func() (goja.Value, error) { - return invokeExtensionMethod(vm, "download", trackID, quality, outputPath, progressCallback) + return invokeExtensionMethod( + vm, + "download", + trackID, + quality, + outputPath, + progressCallback, + downloadOptions, + ) }, ExtDownloadTimeout) perf.recordJS(time.Since(jsStartedAt)) perf.recordPayload(result) diff --git a/go_backend/extension_runtime.go b/go_backend/extension_runtime.go index 4fd0b91e..4e13e216 100644 --- a/go_backend/extension_runtime.go +++ b/go_backend/extension_runtime.go @@ -592,6 +592,7 @@ func (r *extensionRuntime) RegisterAPIs(vm *goja.Runtime) { if r.manifest != nil && r.manifest.Permissions.File { fileObj := vm.NewObject() fileObj.Set("download", r.fileDownload) + fileObj.Set("downloadSegments", r.fileDownloadSegments) fileObj.Set("exists", r.fileExists) fileObj.Set("delete", r.fileDelete) fileObj.Set("read", r.fileRead) diff --git a/go_backend/extension_runtime_chunked.go b/go_backend/extension_runtime_chunked.go new file mode 100644 index 00000000..852ad362 --- /dev/null +++ b/go_backend/extension_runtime_chunked.go @@ -0,0 +1,592 @@ +package gobackend + +import ( + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "time" + + "github.com/dop251/goja" +) + +func chunkedTransferTotal(resp *http.Response) int64 { + if resp == nil { + return 0 + } + if contentRange := resp.Header.Get("Content-Range"); contentRange != "" { + if slash := strings.LastIndex(contentRange, "/"); slash >= 0 { + var total int64 + if _, err := fmt.Sscanf(contentRange[slash+1:], "%d", &total); err == nil { + return total + } + } + } + if resp.StatusCode == http.StatusOK { + return resp.ContentLength + } + return 0 +} + +func (r *extensionRuntime) chunkedTransferCancelled(activeItemID string) bool { + return activeItemID != "" && isDownloadCancelled(activeItemID) +} + +// fileDownloadChunked downloads sequential byte ranges. The same transfer +// policy used by ordinary and segmented downloads controls retries and +// checkpoints here, so specialized CDN downloads do not lose the reliability +// guarantees of the generic file API. +func (r *extensionRuntime) fileDownloadChunked( + client *http.Client, + urlStr, fullPath string, + headers map[string]string, + userAgent string, + chunkSize int64, + onProgress goja.Callable, + trackItemBytes bool, + persistentCheckpoint bool, + policy DownloadTransferPolicy, +) goja.Value { + unlock := lockDownloadOutputPath(fullPath) + defer unlock() + + activeItemID := r.getActiveDownloadItemID() + if activeItemID != "" { + SetItemDownloading(activeItemID) + } + config := transferRetryConfig(policy) + var probeResp *http.Response + var probeFailure transferFailure + probeDelay := config.InitialDelay + probeAttempts := 0 + for attempt := 1; attempt <= policy.MaxAttempts; attempt++ { + probeAttempts = attempt + request, err := http.NewRequest("GET", urlStr, nil) + if err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "invalid_request", + Message: fmt.Sprintf("chunked probe request: %v", err), + Attempts: attempt, + }) + } + request = r.bindDownloadCancelContext(request) + retryContext := request.Context() + for key, value := range headers { + if !strings.EqualFold(key, "Range") { + request.Header.Set(key, value) + } + } + request.Header.Set("User-Agent", userAgent) + request.Header.Set("Range", "bytes=0-1") + request, watchdog := bindStallWatchdog(request, downloadStallTimeout) + response, requestErr := client.Do(request) + if requestErr != nil { + stalled := watchdog.stalled.Load() + watchdog.stop() + if r.chunkedTransferCancelled(activeItemID) { + return r.jsTransferError(transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }) + } + message := fmt.Sprintf("chunked probe failed: %v", requestErr) + if stalled { + message = fmt.Sprintf( + "chunked probe stalled for %ds", + int(downloadStallTimeout.Seconds()), + ) + } + probeFailure = transferFailure{ + ErrorType: "transient_network", + Message: message, + Attempts: attempt, + } + if attempt == policy.MaxAttempts { + return r.jsTransferError(probeFailure) + } + if waitTransferRetry(retryContext, probeDelay) != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }) + } + probeDelay = calculateNextDelay(probeDelay, config) + continue + } + watchdog.stop() + if response.StatusCode == http.StatusPartialContent || + response.StatusCode == http.StatusOK { + io.Copy(io.Discard, io.LimitReader(response.Body, 32*1024)) + response.Body.Close() + probeResp = response + break + } + + retryAfter := retryAfterSeconds(response) + io.Copy(io.Discard, io.LimitReader(response.Body, 32*1024)) + response.Body.Close() + probeFailure = transferFailure{ + ErrorType: transferErrorTypeForStatus(response.StatusCode, policy), + Message: fmt.Sprintf("chunked probe HTTP %d", response.StatusCode), + HTTPStatus: response.StatusCode, + RetryAfterSeconds: retryAfter, + Attempts: attempt, + } + if !retryableTransferStatus(response.StatusCode) || attempt == policy.MaxAttempts { + return r.jsTransferError(probeFailure) + } + delay := probeDelay + if retryAfter > 0 { + delay = time.Duration(retryAfter) * time.Second + } + if waitTransferRetry(retryContext, delay) != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }) + } + probeDelay = calculateNextDelay(probeDelay, config) + } + if probeResp == nil { + return r.jsTransferError(probeFailure) + } + + totalSize := chunkedTransferTotal(probeResp) + validator := transferResponseValidator(probeResp.Header) + fingerprint := transferURLFingerprint(urlStr) + stagedPath := stagedDownloadPath(fullPath) + checkpointPath := transferCheckpointPath(stagedPath) + keepPartial := persistentCheckpoint && validator != "" && fingerprint != "" + checkpoint, checkpointOK := loadTransferCheckpoint(checkpointPath, fingerprint) + if checkpointOK && (checkpoint.Validator != validator || + (checkpoint.Total > 0 && totalSize > 0 && checkpoint.Total != totalSize)) { + checkpointOK = false + } + if !keepPartial || !checkpointOK { + os.Remove(stagedPath) + os.Remove(checkpointPath) + checkpoint = transferCheckpoint{} + } + + output, err := os.OpenFile(stagedPath, os.O_CREATE|os.O_RDWR, 0600) + if err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to create chunked staged file: %v", err), + }) + } + promoted := false + defer func() { + output.Close() + if promoted { + os.Remove(checkpointPath) + } else if !keepPartial { + os.Remove(stagedPath) + os.Remove(checkpointPath) + } + }() + + var totalWritten int64 + if checkpointOK { + if info, statErr := output.Stat(); statErr == nil { + totalWritten = min(checkpoint.Bytes, info.Size()) + } + } + if err := output.Truncate(totalWritten); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to restore chunked partial: %v", err), + }) + } + if _, err := output.Seek(totalWritten, io.SeekStart); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek chunked partial: %v", err), + }) + } + + shouldTrackBytes := activeItemID != "" && trackItemBytes + if shouldTrackBytes { + if totalSize > 0 { + SetItemProgress( + activeItemID, + float64(totalWritten)/float64(totalSize), + totalWritten, + totalSize, + ) + } else if totalWritten > 0 { + SetItemBytesReceived(activeItemID, totalWritten) + } + } + + lastProgressNotify := totalWritten + lastCheckpointBytes := totalWritten + lastCheckpointAt := time.Now() + attemptsUsed := probeAttempts + fullResponse := false + buffer := make([]byte, 64*1024) + for totalSize <= 0 || totalWritten < totalSize { + chunkStart := totalWritten + chunkEnd := chunkStart + chunkSize - 1 + if totalSize > 0 && chunkEnd >= totalSize { + chunkEnd = totalSize - 1 + } + retryDelay := config.InitialDelay + var chunkComplete bool + var lastFailure transferFailure + + for attempt := 1; attempt <= policy.MaxAttempts; attempt++ { + attemptsUsed++ + request, requestErr := http.NewRequest("GET", urlStr, nil) + if requestErr != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "invalid_request", + Message: fmt.Sprintf("chunked request at %d: %v", chunkStart, requestErr), + Attempts: attemptsUsed, + }) + } + request = r.bindDownloadCancelContext(request) + retryContext := request.Context() + for key, value := range headers { + if !strings.EqualFold(key, "Range") { + request.Header.Set(key, value) + } + } + request.Header.Set("User-Agent", userAgent) + request.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", chunkStart, chunkEnd)) + if validator != "" { + request.Header.Set("If-Range", validator) + } + request, watchdog := bindStallWatchdog(request, downloadStallTimeout) + response, responseErr := client.Do(request) + if responseErr != nil { + stalled := watchdog.stalled.Load() + watchdog.stop() + if r.chunkedTransferCancelled(activeItemID) { + lastFailure = transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attemptsUsed, + } + return r.jsTransferError(lastFailure) + } + message := fmt.Sprintf("chunked request at %d failed: %v", chunkStart, responseErr) + if stalled { + message = fmt.Sprintf( + "chunked request at %d stalled for %ds", + chunkStart, + int(downloadStallTimeout.Seconds()), + ) + } + lastFailure = transferFailure{ + ErrorType: "transient_network", + Message: message, + Attempts: attemptsUsed, + } + if attempt == policy.MaxAttempts { + break + } + if waitTransferRetry(retryContext, retryDelay) != nil { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + return r.jsTransferError(lastFailure) + } + retryDelay = calculateNextDelay(retryDelay, config) + continue + } + + if response.StatusCode != http.StatusPartialContent && + response.StatusCode != http.StatusOK { + retryAfter := retryAfterSeconds(response) + io.Copy(io.Discard, io.LimitReader(response.Body, 32*1024)) + response.Body.Close() + watchdog.stop() + lastFailure = transferFailure{ + ErrorType: transferErrorTypeForStatus(response.StatusCode, policy), + Message: fmt.Sprintf("chunked HTTP %d at offset %d", response.StatusCode, chunkStart), + HTTPStatus: response.StatusCode, + RetryAfterSeconds: retryAfter, + Attempts: attemptsUsed, + } + if !retryableTransferStatus(response.StatusCode) || attempt == policy.MaxAttempts { + break + } + delay := retryDelay + if retryAfter > 0 { + delay = time.Duration(retryAfter) * time.Second + } + if waitTransferRetry(retryContext, delay) != nil { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + return r.jsTransferError(lastFailure) + } + retryDelay = calculateNextDelay(retryDelay, config) + continue + } + + if response.StatusCode == http.StatusPartialContent && + !strings.HasPrefix( + response.Header.Get("Content-Range"), + fmt.Sprintf("bytes %d-", chunkStart), + ) { + contentRange := response.Header.Get("Content-Range") + response.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: fmt.Sprintf( + "chunked response has unexpected Content-Range %q at %d", + contentRange, + chunkStart, + ), + HTTPStatus: response.StatusCode, + Attempts: attemptsUsed, + }) + } + if nextValidator := transferResponseValidator(response.Header); nextValidator != "" && + validator != "" && nextValidator != validator { + response.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: "chunked response validator changed during transfer", + Attempts: attemptsUsed, + }) + } + + if response.StatusCode == http.StatusOK && chunkStart > 0 { + if err := output.Truncate(0); err != nil { + response.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to restart ignored range: %v", err), + Attempts: attemptsUsed, + }) + } + if _, err := output.Seek(0, io.SeekStart); err != nil { + response.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek restarted range: %v", err), + Attempts: attemptsUsed, + }) + } + chunkStart = 0 + totalWritten = 0 + lastCheckpointBytes = 0 + os.Remove(checkpointPath) + if response.ContentLength > 0 { + totalSize = response.ContentLength + } + } + + chunkWritten := int64(0) + var readErr error + for { + readCount, bodyErr := response.Body.Read(buffer) + if readCount > 0 { + watchdog.reset() + if r.chunkedTransferCancelled(activeItemID) { + readErr = ErrDownloadCancelled + break + } + writeCount, writeErr := output.Write(buffer[:readCount]) + chunkWritten += int64(writeCount) + totalWritten += int64(writeCount) + if writeErr != nil || writeCount != readCount { + response.Body.Close() + watchdog.stop() + if writeErr == nil { + writeErr = io.ErrShortWrite + } + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to write chunked output: %v", writeErr), + Attempts: attemptsUsed, + }) + } + if shouldTrackBytes { + if totalSize > 0 { + SetItemProgress( + activeItemID, + float64(totalWritten)/float64(totalSize), + totalWritten, + totalSize, + ) + } else { + SetItemBytesReceived(activeItemID, totalWritten) + } + } + if onProgress != nil && totalSize > 0 && + (totalWritten-lastProgressNotify >= progressUpdateThreshold || totalWritten >= totalSize) { + lastProgressNotify = totalWritten + _, _ = onProgress( + goja.Undefined(), + r.vm.ToValue(totalWritten), + r.vm.ToValue(totalSize), + ) + } + } + if bodyErr != nil { + if bodyErr != io.EOF { + readErr = bodyErr + } + break + } + } + response.Body.Close() + stalled := watchdog.stalled.Load() + watchdog.stop() + expectedBytes := response.ContentLength + if response.StatusCode == http.StatusPartialContent && expectedBytes <= 0 { + expectedBytes = chunkEnd - chunkStart + 1 + } + if readErr == nil && expectedBytes > 0 && chunkWritten != expectedBytes { + readErr = io.ErrUnexpectedEOF + } + if readErr == nil && chunkWritten > 0 { + chunkComplete = true + fullResponse = response.StatusCode == http.StatusOK + break + } + + if err := output.Truncate(chunkStart); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to roll back incomplete chunk: %v", err), + Attempts: attemptsUsed, + }) + } + if _, err := output.Seek(chunkStart, io.SeekStart); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek rolled-back chunk: %v", err), + Attempts: attemptsUsed, + }) + } + totalWritten = chunkStart + if shouldTrackBytes && totalSize > 0 { + SetItemProgress( + activeItemID, + float64(totalWritten)/float64(totalSize), + totalWritten, + totalSize, + ) + } + message := fmt.Sprintf("failed to read chunk at %d: %v", chunkStart, readErr) + if chunkWritten == 0 && readErr == nil { + message = fmt.Sprintf("chunk at %d was empty", chunkStart) + } + if stalled { + message = fmt.Sprintf( + "chunk at %d stalled for %ds", + chunkStart, + int(downloadStallTimeout.Seconds()), + ) + } + lastFailure = transferFailure{ + ErrorType: "transient_network", + Message: message, + Attempts: attemptsUsed, + } + if attempt == policy.MaxAttempts || r.chunkedTransferCancelled(activeItemID) { + if r.chunkedTransferCancelled(activeItemID) { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + } + break + } + if waitTransferRetry(retryContext, retryDelay) != nil { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + return r.jsTransferError(lastFailure) + } + retryDelay = calculateNextDelay(retryDelay, config) + } + + if !chunkComplete { + if keepPartial && totalWritten > 0 { + _ = output.Sync() + _ = saveTransferCheckpoint(checkpointPath, transferCheckpoint{ + Fingerprint: fingerprint, + Validator: validator, + Bytes: totalWritten, + Total: totalSize, + }) + } + return r.jsTransferError(lastFailure) + } + if keepPartial && validator != "" && + (totalWritten-lastCheckpointBytes >= transferCheckpointBytes || + time.Since(lastCheckpointAt) >= transferCheckpointPeriod) { + if syncErr := output.Sync(); syncErr == nil { + if saveTransferCheckpoint(checkpointPath, transferCheckpoint{ + Fingerprint: fingerprint, + Validator: validator, + Bytes: totalWritten, + Total: totalSize, + }) == nil { + lastCheckpointBytes = totalWritten + lastCheckpointAt = time.Now() + } + } + } + if fullResponse { + break + } + if totalSize <= 0 && totalWritten-chunkStart < chunkSize { + break + } + } + + if totalWritten <= 0 || (totalSize > 0 && totalWritten != totalSize) { + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: fmt.Sprintf( + "chunked transfer size mismatch: expected %d bytes, wrote %d", + totalSize, + totalWritten, + ), + Attempts: attemptsUsed, + }) + } + if err := output.Sync(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to sync chunked output: %v", err), + Attempts: attemptsUsed, + }) + } + if err := output.Close(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to close chunked output: %v", err), + Attempts: attemptsUsed, + }) + } + if err := os.Rename(stagedPath, fullPath); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to publish chunked output: %v", err), + Attempts: attemptsUsed, + }) + } + promoted = true + os.Remove(checkpointPath) + syncDir(filepath.Dir(fullPath)) + if shouldTrackBytes { + SetItemProgress(activeItemID, 1, totalWritten, totalWritten) + } + return r.jsSuccess(map[string]any{ + "path": fullPath, + "size": totalWritten, + "attempts": attemptsUsed, + }) +} diff --git a/go_backend/extension_runtime_file.go b/go_backend/extension_runtime_file.go index 8c6de5ce..2067e3db 100644 --- a/go_backend/extension_runtime_file.go +++ b/go_backend/extension_runtime_file.go @@ -3,12 +3,10 @@ package gobackend import ( "fmt" "io" - "net/http" "os" "path/filepath" "strings" "sync" - "time" "github.com/dop251/goja" ) @@ -130,6 +128,9 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value { var headers map[string]string var chunkedDownload bool var resumeDownload bool + var resumeOptionSet bool + var persistentCheckpointOption *bool + var maxAttemptsOption int trackItemBytes := true var chunkSize int64 if len(call.Arguments) > 2 && !goja.IsUndefined(call.Arguments[2]) && !goja.IsNull(call.Arguments[2]) { @@ -174,6 +175,20 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value { if resume, ok := opts["resume"]; ok { if v, ok := resume.(bool); ok { resumeDownload = v + resumeOptionSet = true + } + } + if checkpoint, ok := opts["persistentCheckpoint"]; ok { + if v, ok := checkpoint.(bool); ok { + persistentCheckpointOption = &v + } + } + if attempts, ok := opts["maxAttempts"]; ok { + switch v := attempts.(type) { + case int64: + maxAttemptsOption = int(v) + case float64: + maxAttemptsOption = int(v) } } } @@ -199,503 +214,42 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value { ua = h } + policy := r.manifest.DownloadTransferPolicy() + if !resumeOptionSet { + resumeDownload = policy.ResumePolicy == "validated" + } + persistentCheckpoint := policy.PersistentCheckpoint + if persistentCheckpointOption != nil { + persistentCheckpoint = *persistentCheckpointOption && resumeDownload + } + if maxAttemptsOption > 0 { + policy.MaxAttempts = clampInt(maxAttemptsOption, 1, 8) + } if chunkedDownload { - return r.fileDownloadChunked(client, urlStr, fullPath, headers, ua, chunkSize, onProgress, trackItemBytes) + return r.fileDownloadChunked( + client, + urlStr, + fullPath, + headers, + ua, + chunkSize, + onProgress, + trackItemBytes, + persistentCheckpoint, + policy, + ) } - - unlock := lockDownloadOutputPath(fullPath) - defer unlock() - - buildReq := func(rangeFrom int64, validator string) (*http.Request, *stallWatchdog, error) { - req, err := http.NewRequest("GET", urlStr, nil) - if err != nil { - return nil, nil, err - } - req = r.bindDownloadCancelContext(req) - req, wd := bindStallWatchdog(req, downloadStallTimeout) - for k, v := range headers { - req.Header.Set(k, v) - } - if req.Header.Get("User-Agent") == "" { - req.Header.Set("User-Agent", appUserAgent()) - } - if rangeFrom > 0 { - req.Header.Set("Range", fmt.Sprintf("bytes=%d-", rangeFrom)) - req.Header.Set("If-Range", validator) - } - return req, wd, nil - } - - req, wd, err := buildReq(0, "") - if err != nil { - return r.jsError("%s", err.Error()) - } - defer func() { wd.stop() }() - - resp, err := client.Do(req) - if err != nil { - if wd.stalled.Load() { - return r.stallError() - } - return r.jsError("%s", err.Error()) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - resp.Body.Close() - return r.jsError("HTTP error: %d", resp.StatusCode) - } - - // Stream into a staged sibling and promote via rename on success so a - // killed process can never leave a partial file under the final name - // (the duplicate check would then accept it as complete forever). - stagedPath := stagedDownloadPath(fullPath) - os.Remove(stagedPath) - out, err := os.Create(stagedPath) - if err != nil { - resp.Body.Close() - return r.jsError("failed to create file: %v", err) - } - promoted := false - defer func() { - out.Close() - if !promoted { - os.Remove(stagedPath) - } - }() - - activeItemID := r.getActiveDownloadItemID() - if activeItemID != "" { - SetItemDownloading(activeItemID) - } - - contentLength := resp.ContentLength - shouldTrackItemBytes := activeItemID != "" && trackItemBytes - if shouldTrackItemBytes && contentLength > 0 { - SetItemBytesTotal(activeItemID, contentLength) - } - - makeProgressWriter := func() interface{ Write([]byte) (int, error) } { - if shouldTrackItemBytes { - return NewItemProgressWriter(out, activeItemID) - } - return out - } - progressWriter := makeProgressWriter() - - // resumeValidator picks a validator usable with If-Range: a strong ETag or - // Last-Modified. Weak ETags (W/...) are not valid for If-Range. - resumeValidator := func(h http.Header) string { - if etag := h.Get("ETag"); etag != "" && !strings.HasPrefix(etag, "W/") { - return etag - } - return h.Get("Last-Modified") - } - - var written int64 - var lastProgressNotify int64 - buf := make([]byte, 32*1024) - - // copyBody streams resp into progressWriter. fatal is a terminal JS error - // (write failure or user cancel); readErr is a network error eligible for - // a Range resume. - copyBody := func(resp *http.Response) (fatal goja.Value, readErr error) { - defer resp.Body.Close() - for { - nr, er := resp.Body.Read(buf) - if nr > 0 { - wd.reset() - nw, ew := progressWriter.Write(buf[0:nr]) - if nw < 0 || nr < nw { - nw = 0 - if ew == nil { - ew = fmt.Errorf("invalid write result") - } - } - written += int64(nw) - if ew != nil { - if ew == ErrDownloadCancelled { - return r.jsError("download cancelled"), nil - } - return r.jsError("failed to write file: %v", ew), nil - } - if nr != nw { - return r.jsError("short write"), nil - } - - // Throttle the JS callback like the native progress writer: - // per-read invocation is interpreter work inside the copy loop. - if onProgress != nil && contentLength > 0 && - (written-lastProgressNotify >= progressUpdateThreshold || written >= contentLength) { - lastProgressNotify = written - _, _ = onProgress(goja.Undefined(), r.vm.ToValue(written), r.vm.ToValue(contentLength)) - } - } - if er != nil { - if er != io.EOF { - return nil, er - } - return nil, nil - } - } - } - - // Mid-body resume is opt-in because switching networks can route a stable - // URL to a different CDN object even when its validator is unchanged. The - // safe default is to fail and delete the staged partial file. Extensions - // that know their origin supports byte-identical Range resumes can request - // it explicitly with { resume: true }. - validator := resumeValidator(resp.Header) - _, callerSetRange := headers["Range"] - canResume := resumeDownload && validator != "" && !callerSetRange - - const maxResumes = 3 - resumes := 0 - for { - fatal, readErr := copyBody(resp) - if fatal != nil { - return fatal - } - if readErr == nil { - if contentLength <= 0 || written == contentLength { - break - } - // The body reported a clean EOF (e.g. a mid-transfer connection - // reset that the transport surfaced as normal end-of-stream - // instead of io.ErrUnexpectedEOF) but fewer bytes arrived than - // the server promised in Content-Length. Route it through the - // same resume-or-fail path as a real read error instead of - // silently promoting a truncated file. - readErr = io.ErrUnexpectedEOF - } - - stalled := wd.stalled.Load() - if !canResume || resumes >= maxResumes || - (activeItemID != "" && isDownloadCancelled(activeItemID)) { - if stalled { - return r.stallError() - } - return r.jsError("failed to read response: %v", readErr) - } - resumes++ - GoLog("[Extension:%s] Download interrupted at %d bytes (%v), resuming (attempt %d/%d)\n", - r.extensionID, written, readErr, resumes, maxResumes) - - wd.stop() - var resumeReq *http.Request - resumeReq, wd, err = buildReq(written, validator) - if err != nil { - return r.jsError("%s", err.Error()) - } - resp, err = client.Do(resumeReq) - if err != nil { - if wd.stalled.Load() { - return r.stallError() - } - return r.jsError("resume failed at %d bytes: %v", written, err) - } - - switch resp.StatusCode { - case http.StatusPartialContent: - if written > 0 && !strings.HasPrefix(resp.Header.Get("Content-Range"), fmt.Sprintf("bytes %d-", written)) { - cr := resp.Header.Get("Content-Range") - resp.Body.Close() - return r.jsError("resume failed: unexpected Content-Range %q at %d bytes", cr, written) - } - case http.StatusOK: - // Range ignored or file changed under If-Range: restart from zero - // on the same staged file. - if err := out.Truncate(0); err != nil { - resp.Body.Close() - return r.jsError("failed to restart download: %v", err) - } - if _, err := out.Seek(0, io.SeekStart); err != nil { - resp.Body.Close() - return r.jsError("failed to restart download: %v", err) - } - written = 0 - progressWriter = makeProgressWriter() - if resp.ContentLength > 0 { - contentLength = resp.ContentLength - if shouldTrackItemBytes { - SetItemBytesTotal(activeItemID, contentLength) - } - } - validator = resumeValidator(resp.Header) - canResume = resumeDownload && validator != "" - default: - code := resp.StatusCode - resp.Body.Close() - return r.jsError("resume failed: HTTP %d at %d bytes", code, written) - } - } - - if shouldTrackItemBytes { - if contentLength > 0 { - SetItemProgress(activeItemID, float64(written)/float64(contentLength), written, contentLength) - } else if written > 0 { - SetItemBytesReceived(activeItemID, written) - } - } - - // Sync before the promote rename so a power loss right after the rename - // cannot leave a truncated file under the final name. - if err := out.Sync(); err != nil { - return r.jsError("failed to sync file: %v", err) - } - if err := out.Close(); err != nil { - return r.jsError("failed to finalize file: %v", err) - } - if err := os.Rename(stagedPath, fullPath); err != nil { - return r.jsError("failed to publish file: %v", err) - } - promoted = true - syncDir(filepath.Dir(fullPath)) - - GoLog("[Extension:%s] Downloaded %d bytes to %s\n", r.extensionID, written, fullPath) - - return r.jsSuccess(map[string]any{ - "path": fullPath, - "size": written, - }) -} - -// fileDownloadChunked downloads a URL using sequential Range requests. -// This is needed for servers (like YouTube's googlevideo CDN) that reject -// non-ranged or large-range requests with 403 and require small chunk downloads. -func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, fullPath string, headers map[string]string, ua string, chunkSize int64, onProgress goja.Callable, trackItemBytes bool) goja.Value { - unlock := lockDownloadOutputPath(fullPath) - defer unlock() - - // First, get the total content length with a small probe request - probeReq, err := http.NewRequest("GET", urlStr, nil) - if err != nil { - return r.jsError("chunked: probe request error: %v", err) - } - probeReq = r.bindDownloadCancelContext(probeReq) - probeReq, wd := bindStallWatchdog(probeReq, downloadStallTimeout) - defer wd.stop() - stallCtx := probeReq.Context() - probeReq.Header.Set("User-Agent", ua) - for k, v := range headers { - if k != "Range" { // Don't copy any existing Range header - probeReq.Header.Set(k, v) - } - } - probeReq.Header.Set("Range", "bytes=0-1") - - probeResp, err := client.Do(probeReq) - if err != nil { - if wd.stalled.Load() { - return r.stallError() - } - return r.jsError("chunked: probe error: %v", err) - } - io.Copy(io.Discard, probeResp.Body) - probeResp.Body.Close() - - if probeResp.StatusCode != 206 && probeResp.StatusCode != 200 { - return r.jsError("chunked: probe HTTP %d", probeResp.StatusCode) - } - - // Parse Content-Range to get total size: "bytes 0-1/TOTAL" - var totalSize int64 - contentRange := probeResp.Header.Get("Content-Range") - if contentRange != "" { - if idx := strings.LastIndex(contentRange, "/"); idx >= 0 { - sizeStr := contentRange[idx+1:] - if sizeStr != "*" { - fmt.Sscanf(sizeStr, "%d", &totalSize) - } - } - } - - if totalSize <= 0 { - // Fallback: try Content-Length from a HEAD-like approach - // If we can't determine size, download with unknown size - GoLog("[Extension:%s] Chunked download: unknown total size, will download until server says done\n", r.extensionID) - } else { - GoLog("[Extension:%s] Chunked download: total size %d bytes, chunk size %d\n", r.extensionID, totalSize, chunkSize) - } - - // Same staged-write-then-promote protocol as fileDownload: never leave a - // partial file under the final name. - stagedPath := stagedDownloadPath(fullPath) - os.Remove(stagedPath) - out, err := os.Create(stagedPath) - if err != nil { - return r.jsError("failed to create file: %v", err) - } - promoted := false - defer func() { - out.Close() - if !promoted { - os.Remove(stagedPath) - } - }() - - activeItemID := r.getActiveDownloadItemID() - if activeItemID != "" { - SetItemDownloading(activeItemID) - } - - shouldTrackItemBytes := activeItemID != "" && trackItemBytes - if shouldTrackItemBytes && totalSize > 0 { - SetItemBytesTotal(activeItemID, totalSize) - } - - var progressWriter interface{ Write([]byte) (int, error) } = out - if shouldTrackItemBytes { - progressWriter = NewItemProgressWriter(out, activeItemID) - } - - var totalWritten int64 - var lastProgressNotify int64 - buf := make([]byte, 32*1024) - maxRetries := 3 - - for offset := int64(0); totalSize <= 0 || offset < totalSize; { - end := offset + chunkSize - 1 - if totalSize > 0 && end >= totalSize { - end = totalSize - 1 - } - - var chunkResp *http.Response - var chunkErr error - - for retry := 0; retry < maxRetries; retry++ { - chunkReq, err := http.NewRequest("GET", urlStr, nil) - if err != nil { - return r.jsError("chunked: request error at offset %d: %v", offset, err) - } - chunkReq = chunkReq.WithContext(stallCtx) - chunkReq.Header.Set("User-Agent", ua) - for k, v := range headers { - if k != "Range" { - chunkReq.Header.Set(k, v) - } - } - chunkReq.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", offset, end)) - - chunkResp, chunkErr = client.Do(chunkReq) - if chunkErr != nil { - if wd.stalled.Load() { - return r.stallError() - } - if retry < maxRetries-1 { - time.Sleep(time.Duration(retry+1) * time.Second) - continue - } - return r.jsError("chunked: error at offset %d after %d retries: %v", offset, maxRetries, chunkErr) - } - - if chunkResp.StatusCode == 206 || chunkResp.StatusCode == 200 { - break // Success - } - - io.Copy(io.Discard, chunkResp.Body) - chunkResp.Body.Close() - - if chunkResp.StatusCode == 403 || chunkResp.StatusCode == 429 { - if retry < maxRetries-1 { - time.Sleep(time.Duration(retry+1) * 2 * time.Second) - continue - } - } - - return r.jsError("chunked: HTTP %d at offset %d", chunkResp.StatusCode, offset) - } - - chunkWritten := int64(0) - for { - nr, er := chunkResp.Body.Read(buf) - if nr > 0 { - wd.reset() - nw, ew := progressWriter.Write(buf[0:nr]) - if nw < 0 || nr < nw { - nw = 0 - if ew == nil { - ew = fmt.Errorf("invalid write result") - } - } - chunkWritten += int64(nw) - totalWritten += int64(nw) - if ew != nil { - chunkResp.Body.Close() - if ew == ErrDownloadCancelled { - return r.jsError("download cancelled") - } - return r.jsError("failed to write file: %v", ew) - } - if nr != nw { - chunkResp.Body.Close() - return r.jsError("short write") - } - - if onProgress != nil && totalSize > 0 && - (totalWritten-lastProgressNotify >= progressUpdateThreshold || totalWritten >= totalSize) { - lastProgressNotify = totalWritten - _, _ = onProgress(goja.Undefined(), r.vm.ToValue(totalWritten), r.vm.ToValue(totalSize)) - } - } - if er != nil { - if er != io.EOF { - chunkResp.Body.Close() - if wd.stalled.Load() { - return r.stallError() - } - return r.jsError("failed to read chunk at offset %d: %v", offset, er) - } - break - } - } - chunkResp.Body.Close() - - offset += chunkWritten - - // If server returned 200 (full content) instead of 206, we're done - if chunkResp.StatusCode == 200 { - break - } - - // If we got less data than expected and we know total size, check if done - if totalSize > 0 && offset >= totalSize { - break - } - - // Unknown size: if we got less than chunk size, assume done - if totalSize <= 0 && chunkWritten < chunkSize { - break - } - } - - if shouldTrackItemBytes { - if totalSize > 0 { - SetItemProgress(activeItemID, float64(totalWritten)/float64(totalSize), totalWritten, totalSize) - } else if totalWritten > 0 { - SetItemBytesReceived(activeItemID, totalWritten) - } - } - - // Sync before the promote rename so a power loss right after the rename - // cannot leave a truncated file under the final name. - if err := out.Sync(); err != nil { - return r.jsError("failed to sync file: %v", err) - } - if err := out.Close(); err != nil { - return r.jsError("failed to finalize file: %v", err) - } - if err := os.Rename(stagedPath, fullPath); err != nil { - return r.jsError("failed to publish file: %v", err) - } - promoted = true - syncDir(filepath.Dir(fullPath)) - - GoLog("[Extension:%s] Chunked download complete: %d bytes to %s\n", r.extensionID, totalWritten, fullPath) - - return r.jsSuccess(map[string]any{ - "path": fullPath, - "size": totalWritten, - }) + return r.reliableFileDownload( + client, + urlStr, + fullPath, + headers, + onProgress, + trackItemBytes, + resumeDownload, + persistentCheckpoint, + policy, + ) } func (r *extensionRuntime) fileExists(call goja.FunctionCall) goja.Value { diff --git a/go_backend/extension_runtime_file_download_integrity_test.go b/go_backend/extension_runtime_file_download_integrity_test.go index 1bfd07bb..fe959833 100644 --- a/go_backend/extension_runtime_file_download_integrity_test.go +++ b/go_backend/extension_runtime_file_download_integrity_test.go @@ -6,6 +6,7 @@ import ( "net/http" "os" "path/filepath" + "strings" "testing" "github.com/dop251/goja" @@ -33,8 +34,10 @@ func (f *shortCleanEOFBodyReader) Read(p []byte) (int, error) { func TestFileDownloadShortCleanEOFFailsByDefaultEvenWithValidator(t *testing.T) { var attempts int + var rangeSeen bool runtime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { attempts++ + rangeSeen = rangeSeen || req.Header.Get("Range") != "" h := make(http.Header) h.Set("ETag", `"v1"`) return &http.Response{ @@ -46,9 +49,10 @@ func TestFileDownloadShortCleanEOFFailsByDefaultEvenWithValidator(t *testing.T) }, nil }) - // Resume is opt-in (see fileDownload's canResume comment), so a short - // clean EOF must fail and clean up just like a real read error would, - // not silently resume just because a validator happens to be present. + // Resume is opt-in, so a short clean EOF must fail and clean up just like + // a real read error would. The generic transfer engine may retry the whole + // object, but must not send a Range request merely because a validator + // happens to be present. result := runtime.fileDownload(goja.FunctionCall{Arguments: []goja.Value{ runtime.vm.ToValue("https://cdn.example.com/track.flac"), runtime.vm.ToValue("out/track.flac"), @@ -56,8 +60,11 @@ func TestFileDownloadShortCleanEOFFailsByDefaultEvenWithValidator(t *testing.T) if result["success"] != false { t.Fatalf("expected failed download, got %#v", result) } - if attempts != 1 { - t.Fatalf("attempts = %d, want no automatic resume", attempts) + if attempts != defaultTransferMaxAttempts { + t.Fatalf("attempts = %d, want %d full retries", attempts, defaultTransferMaxAttempts) + } + if rangeSeen { + t.Fatal("default retry unexpectedly sent a Range request") } finalPath := filepath.Join(runtime.dataDir, "out", "track.flac") @@ -118,6 +125,65 @@ func TestFileDownloadResumesAfterShortCleanEOFWhenEnabled(t *testing.T) { } } +func TestFileDownloadResumesPersistentCheckpointInNewRuntime(t *testing.T) { + const full = "hello-world!" + firstRuntime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { + header := make(http.Header) + header.Set("ETag", `"v1"`) + return &http.Response{ + StatusCode: http.StatusOK, + Header: header, + Body: io.NopCloser(&shortCleanEOFBodyReader{data: []byte(full[:6])}), + ContentLength: int64(len(full)), + Request: req, + }, nil + }) + options := map[string]any{ + "resume": true, + "persistentCheckpoint": true, + "maxAttempts": float64(1), + } + firstResult := firstRuntime.fileDownload(goja.FunctionCall{Arguments: []goja.Value{ + firstRuntime.vm.ToValue("https://cdn.example.com/track.flac?token=old"), + firstRuntime.vm.ToValue("out/track.flac"), + firstRuntime.vm.ToValue(options), + }}).Export().(map[string]any) + if firstResult["success"] != false { + t.Fatalf("first result = %#v", firstResult) + } + + var resumeRange, resumeIfRange string + secondRuntime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { + resumeRange = req.Header.Get("Range") + resumeIfRange = req.Header.Get("If-Range") + header := make(http.Header) + header.Set("Content-Range", fmt.Sprintf("bytes 6-%d/%d", len(full)-1, len(full))) + return &http.Response{ + StatusCode: http.StatusPartialContent, + Header: header, + Body: io.NopCloser(strings.NewReader(full[6:])), + ContentLength: int64(len(full) - 6), + Request: req, + }, nil + }) + secondRuntime.dataDir = firstRuntime.dataDir + secondResult := secondRuntime.fileDownload(goja.FunctionCall{Arguments: []goja.Value{ + secondRuntime.vm.ToValue("https://cdn.example.com/track.flac?token=fresh"), + secondRuntime.vm.ToValue("out/track.flac"), + secondRuntime.vm.ToValue(options), + }}).Export().(map[string]any) + if secondResult["success"] != true { + t.Fatalf("second result = %#v", secondResult) + } + if resumeRange != "bytes=6-" || resumeIfRange != `"v1"` { + t.Fatalf("range=%q if-range=%q", resumeRange, resumeIfRange) + } + data, err := os.ReadFile(filepath.Join(firstRuntime.dataDir, "out", "track.flac")) + if err != nil || string(data) != full { + t.Fatalf("resumed file = %q, err=%v", data, err) + } +} + func TestFileDownloadShortCleanEOFWithoutValidatorFails(t *testing.T) { runtime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { // No ETag/Last-Modified, so the download cannot be resumed and must @@ -147,3 +213,205 @@ func TestFileDownloadShortCleanEOFWithoutValidatorFails(t *testing.T) { t.Fatalf("staged partial file left behind: %v", err) } } + +func TestChunkedDownloadRestartsWhenServerIgnoresLaterRange(t *testing.T) { + const full = "abcdefgh" + var requestedRanges []string + runtime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { + rangeHeader := req.Header.Get("Range") + requestedRanges = append(requestedRanges, rangeHeader) + header := make(http.Header) + header.Set("ETag", `"v1"`) + switch rangeHeader { + case "bytes=0-1": + header.Set("Content-Range", "bytes 0-1/8") + return &http.Response{ + StatusCode: http.StatusPartialContent, + Header: header, + Body: io.NopCloser(strings.NewReader(full[:2])), + ContentLength: 2, + Request: req, + }, nil + case "bytes=0-2": + header.Set("Content-Range", "bytes 0-2/8") + return &http.Response{ + StatusCode: http.StatusPartialContent, + Header: header, + Body: io.NopCloser(strings.NewReader(full[:3])), + ContentLength: 3, + Request: req, + }, nil + default: + // Some CDNs invalidate or ignore Range after the first request. The + // full response must replace the partial file instead of being appended. + return &http.Response{ + StatusCode: http.StatusOK, + Header: header, + Body: io.NopCloser(strings.NewReader(full)), + ContentLength: int64(len(full)), + Request: req, + }, nil + } + }) + + result := runtime.fileDownload(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue("https://cdn.example.com/track.flac"), + runtime.vm.ToValue("out/track.flac"), + runtime.vm.ToValue(map[string]any{"chunked": float64(3)}), + }}).Export().(map[string]any) + if result["success"] != true { + t.Fatalf("chunked result = %#v", result) + } + data, err := os.ReadFile(filepath.Join(runtime.dataDir, "out", "track.flac")) + if err != nil || string(data) != full { + t.Fatalf("chunked file = %q, err=%v", data, err) + } + if got := requestedRanges[len(requestedRanges)-1]; got != "bytes=3-5" { + t.Fatalf("last requested range = %q, all=%v", got, requestedRanges) + } +} + +func TestFileDownloadRejectsChangedValidatorDuringResume(t *testing.T) { + runtime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { + header := make(http.Header) + header.Set("ETag", `"v2"`) + header.Set("Content-Range", "bytes 3-5/6") + return &http.Response{ + StatusCode: http.StatusPartialContent, + Header: header, + Body: io.NopCloser(strings.NewReader("DEF")), + ContentLength: 3, + Request: req, + }, nil + }) + fullPath := filepath.Join(runtime.dataDir, "out", "track.flac") + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + t.Fatal(err) + } + stagedPath := stagedDownloadPath(fullPath) + if err := os.WriteFile(stagedPath, []byte("ABC"), 0600); err != nil { + t.Fatal(err) + } + url := "https://cdn.example.com/track.flac" + if err := saveTransferCheckpoint(transferCheckpointPath(stagedPath), transferCheckpoint{ + Fingerprint: transferURLFingerprint(url), + Validator: `"v1"`, + Bytes: 3, + Total: 6, + }); err != nil { + t.Fatal(err) + } + + result := runtime.fileDownload(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue(url), + runtime.vm.ToValue("out/track.flac"), + runtime.vm.ToValue(map[string]any{ + "resume": true, + "persistentCheckpoint": true, + }), + }}).Export().(map[string]any) + if result["success"] != false || result["error_type"] != "integrity_failed" { + t.Fatalf("changed-validator result = %#v", result) + } + if _, err := os.Stat(fullPath); !os.IsNotExist(err) { + t.Fatalf("changed entity was published: %v", err) + } +} + +func TestFileDownloadCallerRangeDoesNotReuseEngineCheckpoint(t *testing.T) { + var observedRange string + runtime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { + observedRange = req.Header.Get("Range") + header := make(http.Header) + header.Set("Content-Range", "bytes 5-7/8") + return &http.Response{ + StatusCode: http.StatusPartialContent, + Header: header, + Body: io.NopCloser(strings.NewReader("NEW")), + ContentLength: 3, + Request: req, + }, nil + }) + fullPath := filepath.Join(runtime.dataDir, "out", "fragment.bin") + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + t.Fatal(err) + } + stagedPath := stagedDownloadPath(fullPath) + if err := os.WriteFile(stagedPath, []byte("OLD"), 0600); err != nil { + t.Fatal(err) + } + url := "https://cdn.example.com/track.flac" + if err := saveTransferCheckpoint(transferCheckpointPath(stagedPath), transferCheckpoint{ + Fingerprint: transferURLFingerprint(url), + Validator: `"v1"`, + Bytes: 3, + Total: 8, + }); err != nil { + t.Fatal(err) + } + + result := runtime.fileDownload(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue(url), + runtime.vm.ToValue("out/fragment.bin"), + runtime.vm.ToValue(map[string]any{ + "headers": map[string]any{"Range": "bytes=5-7"}, + "resume": true, + "persistentCheckpoint": true, + }), + }}).Export().(map[string]any) + if result["success"] != true { + t.Fatalf("caller-range result = %#v", result) + } + data, err := os.ReadFile(fullPath) + if err != nil || string(data) != "NEW" { + t.Fatalf("caller-range file = %q, err=%v", data, err) + } + if observedRange != "bytes=5-7" { + t.Fatalf("observed Range = %q", observedRange) + } +} + +func TestFileDownloadPromotesFullyCheckpointedStagedFileWithoutNetwork(t *testing.T) { + var networkCalls int + runtime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { + networkCalls++ + return nil, fmt.Errorf("network should not be called") + }) + const full = "already-complete" + fullPath := filepath.Join(runtime.dataDir, "out", "track.flac") + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + t.Fatal(err) + } + stagedPath := stagedDownloadPath(fullPath) + if err := os.WriteFile(stagedPath, []byte(full), 0600); err != nil { + t.Fatal(err) + } + url := "https://cdn.example.com/track.flac" + if err := saveTransferCheckpoint(transferCheckpointPath(stagedPath), transferCheckpoint{ + Fingerprint: transferURLFingerprint(url), + Validator: `"v1"`, + Bytes: int64(len(full)), + Total: int64(len(full)), + }); err != nil { + t.Fatal(err) + } + + result := runtime.fileDownload(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue(url), + runtime.vm.ToValue("out/track.flac"), + runtime.vm.ToValue(map[string]any{ + "resume": true, + "persistentCheckpoint": true, + }), + }}).Export().(map[string]any) + if result["success"] != true || result["resumed"] != true { + t.Fatalf("fully checkpointed result = %#v", result) + } + if networkCalls != 0 { + t.Fatalf("network calls = %d", networkCalls) + } + data, err := os.ReadFile(fullPath) + if err != nil || string(data) != full { + t.Fatalf("promoted file = %q, err=%v", data, err) + } +} diff --git a/go_backend/extension_runtime_segments.go b/go_backend/extension_runtime_segments.go new file mode 100644 index 00000000..12e52cca --- /dev/null +++ b/go_backend/extension_runtime_segments.go @@ -0,0 +1,741 @@ +package gobackend + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/dop251/goja" +) + +type segmentTransferSpec struct { + Index int + URL string + Headers map[string]string +} + +type segmentTransferResult struct { + Index int + Path string + Size int64 + Attempts int + Failure *transferFailure +} + +type segmentTransferCheckpoint struct { + Version int `json:"version"` + Fingerprint string `json:"fingerprint"` + NextIndex int `json:"next_index"` + Bytes int64 `json:"bytes"` + UpdatedAt int64 `json:"updated_at"` +} + +func parseStringHeaders(value any) map[string]string { + raw, ok := value.(map[string]any) + if !ok { + return nil + } + headers := make(map[string]string, len(raw)) + for key, entry := range raw { + headers[key] = fmt.Sprintf("%v", entry) + } + return headers +} + +func mergeStringHeaders(base, override map[string]string) map[string]string { + if len(base) == 0 && len(override) == 0 { + return nil + } + merged := make(map[string]string, len(base)+len(override)) + for key, value := range base { + merged[key] = value + } + for key, value := range override { + merged[key] = value + } + return merged +} + +func parseSegmentTransferSpecs(value any, commonHeaders map[string]string) ([]segmentTransferSpec, error) { + rawSegments, ok := value.([]any) + if !ok || len(rawSegments) == 0 { + return nil, fmt.Errorf("segments must be a non-empty array") + } + segments := make([]segmentTransferSpec, 0, len(rawSegments)) + for index, raw := range rawSegments { + var rawURL string + var headers map[string]string + switch typed := raw.(type) { + case string: + rawURL = typed + case map[string]any: + rawURL, _ = typed["url"].(string) + headers = parseStringHeaders(typed["headers"]) + default: + return nil, fmt.Errorf("segment %d must be a URL string or object", index) + } + rawURL = strings.TrimSpace(rawURL) + if rawURL == "" { + return nil, fmt.Errorf("segment %d URL is empty", index) + } + segments = append(segments, segmentTransferSpec{ + Index: index, + URL: rawURL, + Headers: mergeStringHeaders(commonHeaders, headers), + }) + } + return segments, nil +} + +func segmentListFingerprint(segments []segmentTransferSpec) string { + hash := sha256.New() + for _, segment := range segments { + // Segmented checkpoints have no per-segment ETag or Last-Modified + // validator. Include the complete URL, including its query, so a + // different media object served from the same CDN path can never reuse + // already-assembled bytes. Ordinary single-file checkpoints may ignore + // rotating query credentials because their validator still protects + // integrity. + hash.Write([]byte(segment.URL)) + hash.Write([]byte{0}) + } + return hex.EncodeToString(hash.Sum(nil)) +} + +func loadSegmentCheckpoint(path, fingerprint string) (segmentTransferCheckpoint, bool) { + var checkpoint segmentTransferCheckpoint + data, err := os.ReadFile(path) + if err != nil || json.Unmarshal(data, &checkpoint) != nil { + return segmentTransferCheckpoint{}, false + } + if checkpoint.Version != transferCheckpointVersion || + checkpoint.Fingerprint != fingerprint || + checkpoint.NextIndex < 0 || checkpoint.Bytes < 0 || + (checkpoint.NextIndex == 0 && checkpoint.Bytes != 0) || + (checkpoint.NextIndex > 0 && checkpoint.Bytes == 0) { + return segmentTransferCheckpoint{}, false + } + return checkpoint, true +} + +func saveSegmentCheckpoint(path string, checkpoint segmentTransferCheckpoint) error { + checkpoint.Version = transferCheckpointVersion + checkpoint.UpdatedAt = time.Now().UnixMilli() + data, err := json.Marshal(checkpoint) + if err != nil { + return err + } + tempPath := path + ".tmp" + file, err := os.OpenFile(tempPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600) + if err != nil { + return err + } + if _, err = file.Write(data); err == nil { + err = file.Sync() + } + closeErr := file.Close() + if err == nil { + err = closeErr + } + if err != nil { + os.Remove(tempPath) + return err + } + if err := os.Rename(tempPath, path); err != nil { + os.Remove(tempPath) + return err + } + return nil +} + +func segmentTempPath(stagedPath string, index int) string { + return fmt.Sprintf("%s.segment.%06d", stagedPath, index) +} + +func (r *extensionRuntime) fetchSegmentToTemp( + ctx context.Context, + client *http.Client, + spec segmentTransferSpec, + tempPath string, + policy DownloadTransferPolicy, + received *atomic.Int64, + activeItemID string, +) segmentTransferResult { + config := transferRetryConfig(policy) + retryDelay := config.InitialDelay + var lastFailure transferFailure + + for attempt := 1; attempt <= policy.MaxAttempts; attempt++ { + if ctx.Err() != nil { + return segmentTransferResult{ + Index: spec.Index, + Failure: &transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }, + } + } + os.Remove(tempPath) + output, err := os.OpenFile(tempPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600) + if err != nil { + return segmentTransferResult{ + Index: spec.Index, + Failure: &transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to create segment file: %v", err), + Attempts: attempt, + }, + } + } + + req, err := http.NewRequestWithContext(ctx, "GET", spec.URL, nil) + if err != nil { + output.Close() + return segmentTransferResult{ + Index: spec.Index, + Failure: &transferFailure{ + ErrorType: "invalid_request", + Message: err.Error(), + Attempts: attempt, + }, + } + } + for key, value := range spec.Headers { + req.Header.Set(key, value) + } + if req.Header.Get("User-Agent") == "" { + req.Header.Set("User-Agent", appUserAgent()) + } + req, watchdog := bindStallWatchdog(req, downloadStallTimeout) + resp, err := client.Do(req) + if err != nil { + stalled := watchdog.stalled.Load() + watchdog.stop() + output.Close() + message := err.Error() + if stalled { + message = fmt.Sprintf( + "segment %d stalled for %ds", + spec.Index, + int(downloadStallTimeout.Seconds()), + ) + } + lastFailure = transferFailure{ + ErrorType: "transient_network", + Message: message, + Attempts: attempt, + } + if attempt == policy.MaxAttempts || ctx.Err() != nil { + if ctx.Err() != nil { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + } + return segmentTransferResult{Index: spec.Index, Failure: &lastFailure} + } + if waitTransferRetry(ctx, retryDelay) != nil { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + return segmentTransferResult{Index: spec.Index, Failure: &lastFailure} + } + retryDelay = calculateNextDelay(retryDelay, config) + continue + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + retryAfter := retryAfterSeconds(resp) + io.Copy(io.Discard, io.LimitReader(resp.Body, 32*1024)) + resp.Body.Close() + watchdog.stop() + output.Close() + lastFailure = transferFailure{ + ErrorType: transferErrorTypeForStatus(resp.StatusCode, policy), + Message: fmt.Sprintf("segment %d HTTP error: %d", spec.Index, resp.StatusCode), + HTTPStatus: resp.StatusCode, + RetryAfterSeconds: retryAfter, + Attempts: attempt, + } + if !retryableTransferStatus(resp.StatusCode) || attempt == policy.MaxAttempts { + return segmentTransferResult{Index: spec.Index, Failure: &lastFailure} + } + delay := retryDelay + if retryAfter > 0 { + delay = time.Duration(retryAfter) * time.Second + } + if waitTransferRetry(ctx, delay) != nil { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + return segmentTransferResult{Index: spec.Index, Failure: &lastFailure} + } + retryDelay = calculateNextDelay(retryDelay, config) + continue + } + + buffer := make([]byte, 64*1024) + var size int64 + var readErr error + for { + readCount, bodyErr := resp.Body.Read(buffer) + if readCount > 0 { + watchdog.reset() + writeCount, writeErr := output.Write(buffer[:readCount]) + size += int64(writeCount) + received.Add(int64(writeCount)) + if activeItemID != "" { + SetItemBytesReceived(activeItemID, received.Load()) + } + if writeErr != nil || writeCount != readCount { + if writeErr == nil { + writeErr = io.ErrShortWrite + } + readErr = writeErr + break + } + } + if bodyErr != nil { + if bodyErr != io.EOF { + readErr = bodyErr + } + break + } + } + resp.Body.Close() + stalled := watchdog.stalled.Load() + watchdog.stop() + closeErr := output.Close() + if readErr == nil { + readErr = closeErr + } + if readErr == nil && resp.ContentLength > 0 && size != resp.ContentLength { + readErr = io.ErrUnexpectedEOF + } + if readErr == nil && size > 0 { + return segmentTransferResult{ + Index: spec.Index, + Path: tempPath, + Size: size, + Attempts: attempt, + } + } + if size > 0 { + received.Add(-size) + } + + message := fmt.Sprintf("failed to read segment %d: %v", spec.Index, readErr) + if size == 0 && readErr == nil { + message = fmt.Sprintf("segment %d response was empty", spec.Index) + } + if stalled { + message = fmt.Sprintf( + "segment %d stalled for %ds", + spec.Index, + int(downloadStallTimeout.Seconds()), + ) + } + lastFailure = transferFailure{ + ErrorType: "transient_network", + Message: message, + Attempts: attempt, + } + if attempt == policy.MaxAttempts { + return segmentTransferResult{Index: spec.Index, Failure: &lastFailure} + } + if waitTransferRetry(ctx, retryDelay) != nil { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + return segmentTransferResult{Index: spec.Index, Failure: &lastFailure} + } + retryDelay = calculateNextDelay(retryDelay, config) + } + + return segmentTransferResult{ + Index: spec.Index, + Failure: &transferFailure{ + ErrorType: "transient_network", + Message: fmt.Sprintf("segment %d exhausted retry budget", spec.Index), + Attempts: policy.MaxAttempts, + }, + } +} + +func (r *extensionRuntime) fileDownloadSegments(call goja.FunctionCall) goja.Value { + if len(call.Arguments) < 2 { + return r.jsTransferError(transferFailure{ + ErrorType: "invalid_request", + Message: "segments and output path are required", + }) + } + + var commonHeaders map[string]string + var onProgress goja.Callable + var maxParallelOption int + var maxAttemptsOption int + var persistentCheckpointOption *bool + if len(call.Arguments) > 2 && + !goja.IsUndefined(call.Arguments[2]) && + !goja.IsNull(call.Arguments[2]) { + if options, ok := call.Arguments[2].Export().(map[string]any); ok { + commonHeaders = parseStringHeaders(options["headers"]) + if progressValue, ok := options["onProgress"]; ok { + if callable, ok := goja.AssertFunction(r.vm.ToValue(progressValue)); ok { + onProgress = callable + } + } + maxParallelOption = capabilityInt(options["maxParallel"], 0) + maxAttemptsOption = capabilityInt(options["maxAttempts"], 0) + if checkpoint, ok := options["persistentCheckpoint"].(bool); ok { + persistentCheckpointOption = &checkpoint + } + } + } + + segments, err := parseSegmentTransferSpecs(call.Arguments[0].Export(), commonHeaders) + if err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "invalid_request", + Message: err.Error(), + }) + } + for _, segment := range segments { + if err := r.validateDomain(segment.URL); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "permission", + Message: err.Error(), + }) + } + } + fullPath, err := r.validatePath(call.Arguments[1].String()) + if err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "permission", + Message: err.Error(), + }) + } + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to create output directory: %v", err), + }) + } + + policy := r.manifest.DownloadTransferPolicy() + if maxParallelOption > 0 { + policy.MaxParallelSegments = clampInt(maxParallelOption, 1, maxParallelSegments) + } + if maxAttemptsOption > 0 { + policy.MaxAttempts = clampInt(maxAttemptsOption, 1, 8) + } + persistentCheckpoint := policy.PersistentCheckpoint + if persistentCheckpointOption != nil { + persistentCheckpoint = *persistentCheckpointOption + } + + client := r.downloadClient + if client == nil { + client = r.httpClient + } + unlock := lockDownloadOutputPath(fullPath) + defer unlock() + + stagedPath := stagedDownloadPath(fullPath) + checkpointPath := transferCheckpointPath(stagedPath) + ".segments" + fingerprint := segmentListFingerprint(segments) + checkpoint, checkpointOK := loadSegmentCheckpoint(checkpointPath, fingerprint) + if !persistentCheckpoint || !checkpointOK || checkpoint.NextIndex > len(segments) { + checkpoint = segmentTransferCheckpoint{Fingerprint: fingerprint} + checkpointOK = false + os.Remove(stagedPath) + os.Remove(checkpointPath) + } + + output, err := os.OpenFile(stagedPath, os.O_CREATE|os.O_RDWR, 0600) + if err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to create segmented output: %v", err), + }) + } + promoted := false + defer func() { + output.Close() + for index := range segments { + os.Remove(segmentTempPath(stagedPath, index)) + } + if promoted { + os.Remove(checkpointPath) + } else if !persistentCheckpoint { + os.Remove(stagedPath) + os.Remove(checkpointPath) + } + }() + + nextIndex := 0 + var totalWritten int64 + if checkpointOK { + if info, statErr := output.Stat(); statErr == nil && info.Size() >= checkpoint.Bytes { + totalWritten = checkpoint.Bytes + nextIndex = checkpoint.NextIndex + } else { + // Segment boundaries cannot be reconstructed from a shorter file. + // Restart instead of skipping segments named by a stale checkpoint. + checkpointOK = false + checkpoint = segmentTransferCheckpoint{Fingerprint: fingerprint} + os.Remove(checkpointPath) + } + } + if err := output.Truncate(totalWritten); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to restore segmented output: %v", err), + }) + } + if _, err := output.Seek(totalWritten, io.SeekStart); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek segmented output: %v", err), + }) + } + + activeItemID := r.getActiveDownloadItemID() + if activeItemID != "" { + SetItemDownloading(activeItemID) + SetItemProgress( + activeItemID, + float64(nextIndex)/float64(len(segments)), + totalWritten, + 0, + ) + } + if nextIndex == len(segments) && totalWritten > 0 { + if err := output.Sync(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to sync restored segmented output: %v", err), + }) + } + if err := output.Close(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to close restored segmented output: %v", err), + }) + } + if err := os.Rename(stagedPath, fullPath); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to publish restored segmented output: %v", err), + }) + } + promoted = true + os.Remove(checkpointPath) + syncDir(filepath.Dir(fullPath)) + if activeItemID != "" { + SetItemProgress(activeItemID, 1, totalWritten, totalWritten) + } + return r.jsSuccess(map[string]any{ + "path": fullPath, + "size": totalWritten, + "segments": len(segments), + "resumed": true, + }) + } + + baseRequest, requestErr := http.NewRequest("GET", segments[nextIndex].URL, nil) + if requestErr != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "invalid_request", + Message: requestErr.Error(), + }) + } + baseRequest = r.bindDownloadCancelContext(baseRequest) + ctx, cancel := context.WithCancel(baseRequest.Context()) + defer cancel() + + jobs := make(chan segmentTransferSpec) + results := make(chan segmentTransferResult, policy.MaxParallelSegments) + var received atomic.Int64 + received.Store(totalWritten) + var workers sync.WaitGroup + workerCount := min(policy.MaxParallelSegments, len(segments)-nextIndex) + for workerIndex := 0; workerIndex < workerCount; workerIndex++ { + workers.Add(1) + go func() { + defer workers.Done() + for spec := range jobs { + result := r.fetchSegmentToTemp( + ctx, + client, + spec, + segmentTempPath(stagedPath, spec.Index), + policy, + &received, + activeItemID, + ) + select { + case results <- result: + case <-ctx.Done(): + return + } + if result.Failure != nil { + return + } + } + }() + } + go func() { + defer close(jobs) + for index := nextIndex; index < len(segments); index++ { + select { + case jobs <- segments[index]: + case <-ctx.Done(): + return + } + } + }() + go func() { + workers.Wait() + close(results) + }() + + pending := make(map[int]segmentTransferResult) + completedSegments := nextIndex + lastCheckpointBytes := totalWritten + lastCheckpointAt := time.Now() + var firstFailure *transferFailure + for result := range results { + if result.Failure != nil { + if firstFailure == nil { + failureCopy := *result.Failure + firstFailure = &failureCopy + cancel() + } + continue + } + pending[result.Index] = result + for { + ready, ok := pending[nextIndex] + if !ok { + break + } + segmentFile, openErr := os.Open(ready.Path) + if openErr != nil { + firstFailure = &transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to open downloaded segment %d: %v", nextIndex, openErr), + Attempts: ready.Attempts, + } + cancel() + break + } + copied, copyErr := io.CopyBuffer(output, segmentFile, make([]byte, 128*1024)) + segmentFile.Close() + if copyErr != nil || copied != ready.Size { + if copyErr == nil { + copyErr = io.ErrShortWrite + } + firstFailure = &transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to append segment %d: %v", nextIndex, copyErr), + Attempts: ready.Attempts, + } + cancel() + break + } + totalWritten += copied + os.Remove(ready.Path) + delete(pending, nextIndex) + nextIndex++ + completedSegments++ + if persistentCheckpoint && + (totalWritten-lastCheckpointBytes >= transferCheckpointBytes || + time.Since(lastCheckpointAt) >= transferCheckpointPeriod) { + if output.Sync() == nil && saveSegmentCheckpoint( + checkpointPath, + segmentTransferCheckpoint{ + Fingerprint: fingerprint, + NextIndex: nextIndex, + Bytes: totalWritten, + }, + ) == nil { + lastCheckpointBytes = totalWritten + lastCheckpointAt = time.Now() + } + } + if activeItemID != "" { + SetItemProgress( + activeItemID, + float64(completedSegments)/float64(len(segments)), + received.Load(), + 0, + ) + } + if onProgress != nil { + _, _ = onProgress( + goja.Undefined(), + r.vm.ToValue(received.Load()), + r.vm.ToValue(int64(0)), + r.vm.ToValue(completedSegments), + r.vm.ToValue(len(segments)), + ) + } + } + } + if firstFailure != nil { + if persistentCheckpoint && nextIndex > 0 && totalWritten > 0 && output.Sync() == nil { + _ = saveSegmentCheckpoint(checkpointPath, segmentTransferCheckpoint{ + Fingerprint: fingerprint, + NextIndex: nextIndex, + Bytes: totalWritten, + }) + } + return r.jsTransferError(*firstFailure) + } + if nextIndex != len(segments) || totalWritten <= 0 { + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: fmt.Sprintf( + "segmented transfer incomplete: assembled %d of %d segments", + nextIndex, + len(segments), + ), + }) + } + if err := output.Sync(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to sync segmented output: %v", err), + }) + } + if err := output.Close(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to close segmented output: %v", err), + }) + } + if err := os.Rename(stagedPath, fullPath); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to publish segmented output: %v", err), + }) + } + promoted = true + os.Remove(checkpointPath) + syncDir(filepath.Dir(fullPath)) + if activeItemID != "" { + SetItemProgress(activeItemID, 1, totalWritten, totalWritten) + } + return r.jsSuccess(map[string]any{ + "path": fullPath, + "size": totalWritten, + "segments": len(segments), + }) +} diff --git a/go_backend/extension_runtime_segments_test.go b/go_backend/extension_runtime_segments_test.go new file mode 100644 index 00000000..535b3a7f --- /dev/null +++ b/go_backend/extension_runtime_segments_test.go @@ -0,0 +1,257 @@ +package gobackend + +import ( + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/dop251/goja" +) + +func segmentTestRuntime(t *testing.T, transport roundTripFunc) *extensionRuntime { + t.Helper() + runtime := newFileDownloadTestRuntime(t, transport) + runtime.manifest.Capabilities = map[string]any{ + "downloadTransfer": map[string]any{ + "maxAttempts": float64(3), + "initialRetryDelayMs": float64(100), + "maxRetryDelayMs": float64(100), + "resumePolicy": "validated", + "persistentCheckpoint": true, + "maxParallelSegments": float64(3), + }, + } + return runtime +} + +func TestFileDownloadSegmentsPreservesOrderRetriesAndRunsConcurrently(t *testing.T) { + var active atomic.Int32 + var maxActive atomic.Int32 + var mu sync.Mutex + attempts := map[string]int{} + runtime := segmentTestRuntime(t, func(req *http.Request) (*http.Response, error) { + name := strings.TrimPrefix(req.URL.Path, "/") + mu.Lock() + attempts[name]++ + attempt := attempts[name] + mu.Unlock() + + current := active.Add(1) + defer active.Add(-1) + for { + previous := maxActive.Load() + if current <= previous || maxActive.CompareAndSwap(previous, current) { + break + } + } + time.Sleep(40 * time.Millisecond) + if name == "segment-1" && attempt == 1 { + return &http.Response{ + StatusCode: http.StatusServiceUnavailable, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("retry")), + Request: req, + }, nil + } + body := map[string]string{ + "segment-0": "A", + "segment-1": "B", + "segment-2": "C", + }[name] + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(body)), + ContentLength: int64(len(body)), + Request: req, + }, nil + }) + + segments := []any{ + "https://cdn.example.com/segment-0", + "https://cdn.example.com/segment-1", + "https://cdn.example.com/segment-2", + } + result := runtime.fileDownloadSegments(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue(segments), + runtime.vm.ToValue("out/track.flac"), + }}).Export().(map[string]any) + if result["success"] != true { + t.Fatalf("segmented result = %#v", result) + } + data, err := os.ReadFile(filepath.Join(runtime.dataDir, "out", "track.flac")) + if err != nil || string(data) != "ABC" { + t.Fatalf("assembled data = %q, err=%v", data, err) + } + if maxActive.Load() < 2 { + t.Fatalf("segments did not overlap; max active = %d", maxActive.Load()) + } + mu.Lock() + segmentOneAttempts := attempts["segment-1"] + mu.Unlock() + if segmentOneAttempts != 2 { + t.Fatalf("segment-1 attempts = %d, want 2", segmentOneAttempts) + } +} + +func TestFileDownloadSegmentsResumesAssembledCheckpoint(t *testing.T) { + requested := make(chan string, 2) + runtime := segmentTestRuntime(t, func(req *http.Request) (*http.Response, error) { + name := strings.TrimPrefix(req.URL.Path, "/") + requested <- name + body := "B" + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(body)), + ContentLength: int64(len(body)), + Request: req, + }, nil + }) + + segments := []segmentTransferSpec{ + {Index: 0, URL: "https://cdn.example.com/segment-0"}, + {Index: 1, URL: "https://cdn.example.com/segment-1"}, + } + fullPath := filepath.Join(runtime.dataDir, "out", "track.flac") + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + t.Fatal(err) + } + stagedPath := stagedDownloadPath(fullPath) + if err := os.WriteFile(stagedPath, []byte("A"), 0600); err != nil { + t.Fatal(err) + } + checkpointPath := transferCheckpointPath(stagedPath) + ".segments" + if err := saveSegmentCheckpoint(checkpointPath, segmentTransferCheckpoint{ + Fingerprint: segmentListFingerprint(segments), + NextIndex: 1, + Bytes: 1, + }); err != nil { + t.Fatal(err) + } + + result := runtime.fileDownloadSegments(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue([]any{segments[0].URL, segments[1].URL}), + runtime.vm.ToValue("out/track.flac"), + }}).Export().(map[string]any) + if result["success"] != true { + t.Fatalf("segmented resume result = %#v", result) + } + close(requested) + requests := []string{} + for name := range requested { + requests = append(requests, name) + } + if len(requests) != 1 || requests[0] != "segment-1" { + t.Fatalf("requests after checkpoint = %v", requests) + } + data, err := os.ReadFile(fullPath) + if err != nil || string(data) != "AB" { + t.Fatalf("resumed data = %q, err=%v", data, err) + } + if _, err := os.Stat(checkpointPath); !os.IsNotExist(err) { + t.Fatalf("checkpoint not removed after publish: %v", err) + } +} + +func TestFileDownloadSegmentsReturnsTypedExpiredStreamError(t *testing.T) { + runtime := segmentTestRuntime(t, func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusForbidden, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("expired")), + Request: req, + }, nil + }) + result := runtime.fileDownloadSegments(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue([]any{"https://cdn.example.com/segment-0"}), + runtime.vm.ToValue("out/track.flac"), + }}).Export().(map[string]any) + if result["success"] != false || result["error_type"] != "expired_stream" || + fmt.Sprint(result["http_status"]) != fmt.Sprint(http.StatusForbidden) { + t.Fatalf("typed error = %#v", result) + } + if message := fmt.Sprint(result["error"]); !strings.Contains(message, "403") { + t.Fatalf("typed error message = %q", message) + } +} + +func TestFileDownloadSegmentsRestartsWhenCheckpointExceedsStagedFile(t *testing.T) { + var mu sync.Mutex + requested := []string{} + runtime := segmentTestRuntime(t, func(req *http.Request) (*http.Response, error) { + name := strings.TrimPrefix(req.URL.Path, "/") + mu.Lock() + requested = append(requested, name) + mu.Unlock() + body := map[string]string{"segment-0": "A", "segment-1": "B"}[name] + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(body)), + ContentLength: int64(len(body)), + Request: req, + }, nil + }) + segments := []segmentTransferSpec{ + {Index: 0, URL: "https://cdn.example.com/segment-0"}, + {Index: 1, URL: "https://cdn.example.com/segment-1"}, + } + fullPath := filepath.Join(runtime.dataDir, "out", "track.flac") + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + t.Fatal(err) + } + stagedPath := stagedDownloadPath(fullPath) + if err := os.WriteFile(stagedPath, nil, 0600); err != nil { + t.Fatal(err) + } + if err := saveSegmentCheckpoint( + transferCheckpointPath(stagedPath)+".segments", + segmentTransferCheckpoint{ + Fingerprint: segmentListFingerprint(segments), + NextIndex: 1, + Bytes: 1, + }, + ); err != nil { + t.Fatal(err) + } + + result := runtime.fileDownloadSegments(goja.FunctionCall{Arguments: []goja.Value{ + runtime.vm.ToValue([]any{segments[0].URL, segments[1].URL}), + runtime.vm.ToValue("out/track.flac"), + }}).Export().(map[string]any) + if result["success"] != true { + t.Fatalf("stale-checkpoint result = %#v", result) + } + data, err := os.ReadFile(fullPath) + if err != nil || string(data) != "AB" { + t.Fatalf("restarted segmented data = %q, err=%v", data, err) + } + mu.Lock() + requestCount := len(requested) + mu.Unlock() + if requestCount != 2 { + t.Fatalf("requested segments = %v", requested) + } +} + +func TestSegmentCheckpointFingerprintIncludesQueryIdentity(t *testing.T) { + first := []segmentTransferSpec{{ + Index: 0, + URL: "https://cdn.example.com/audio?media=first", + }} + second := []segmentTransferSpec{{ + Index: 0, + URL: "https://cdn.example.com/audio?media=second", + }} + if segmentListFingerprint(first) == segmentListFingerprint(second) { + t.Fatal("different segment query identities shared a checkpoint fingerprint") + } +} diff --git a/go_backend/extension_runtime_supplement_test.go b/go_backend/extension_runtime_supplement_test.go index d87878ab..72008c13 100644 --- a/go_backend/extension_runtime_supplement_test.go +++ b/go_backend/extension_runtime_supplement_test.go @@ -270,8 +270,10 @@ func TestFileDownloadFailureLeavesNoFinalFile(t *testing.T) { func TestFileDownloadDoesNotResumeMidBodyCutByDefault(t *testing.T) { const full = "hello-world!" var attempts int + var rangeSeen bool runtime := newFileDownloadTestRuntime(t, func(req *http.Request) (*http.Response, error) { attempts++ + rangeSeen = rangeSeen || req.Header.Get("Range") != "" h := make(http.Header) h.Set("ETag", `"v1"`) return &http.Response{ @@ -290,8 +292,11 @@ func TestFileDownloadDoesNotResumeMidBodyCutByDefault(t *testing.T) { if result["success"] != false { t.Fatalf("expected failed download, got %#v", result) } - if attempts != 1 { - t.Fatalf("attempts = %d, want no automatic resume", attempts) + if attempts != defaultTransferMaxAttempts { + t.Fatalf("attempts = %d, want %d full retries", attempts, defaultTransferMaxAttempts) + } + if rangeSeen { + t.Fatal("default retry unexpectedly sent a Range request") } finalPath := filepath.Join(runtime.dataDir, "out", "track.flac") if _, err := os.Stat(finalPath); !os.IsNotExist(err) { diff --git a/go_backend/extension_runtime_transfer.go b/go_backend/extension_runtime_transfer.go new file mode 100644 index 00000000..02b5957c --- /dev/null +++ b/go_backend/extension_runtime_transfer.go @@ -0,0 +1,720 @@ +package gobackend + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/dop251/goja" +) + +const ( + transferCheckpointVersion = 1 + transferCheckpointBytes = 8 * 1024 * 1024 + transferCheckpointPeriod = 5 * time.Second +) + +type transferCheckpoint struct { + Version int `json:"version"` + Fingerprint string `json:"fingerprint"` + Validator string `json:"validator"` + Bytes int64 `json:"bytes"` + Total int64 `json:"total,omitempty"` + UpdatedAt int64 `json:"updated_at"` +} + +type transferFailure struct { + ErrorType string + Message string + HTTPStatus int + RetryAfterSeconds int + Attempts int +} + +func (r *extensionRuntime) jsTransferError(failure transferFailure) goja.Value { + values := map[string]any{ + "success": false, + "error": failure.Message, + "error_type": failure.ErrorType, + "attempts": failure.Attempts, + } + if failure.HTTPStatus > 0 { + values["http_status"] = failure.HTTPStatus + } + if failure.RetryAfterSeconds > 0 { + values["retry_after_seconds"] = failure.RetryAfterSeconds + } + return r.vm.ToValue(values) +} + +func transferURLFingerprint(rawURL string) string { + parsed, err := url.Parse(rawURL) + if err != nil { + return "" + } + // Query strings commonly contain short-lived CDN credentials. Excluding + // them both avoids persisting a secret-derived value and permits a freshly + // signed URL for the same object to continue a validator-protected partial. + identity := strings.ToLower(parsed.Scheme) + "://" + + strings.ToLower(parsed.Host) + parsed.EscapedPath() + sum := sha256.Sum256([]byte(identity)) + return hex.EncodeToString(sum[:]) +} + +func transferCheckpointPath(stagedPath string) string { + return stagedPath + ".checkpoint.json" +} + +func loadTransferCheckpoint(path, fingerprint string) (transferCheckpoint, bool) { + var checkpoint transferCheckpoint + data, err := os.ReadFile(path) + if err != nil || json.Unmarshal(data, &checkpoint) != nil { + return transferCheckpoint{}, false + } + if checkpoint.Version != transferCheckpointVersion || + checkpoint.Fingerprint == "" || + checkpoint.Fingerprint != fingerprint || + checkpoint.Validator == "" || + checkpoint.Bytes <= 0 { + return transferCheckpoint{}, false + } + return checkpoint, true +} + +func saveTransferCheckpoint(path string, checkpoint transferCheckpoint) error { + checkpoint.Version = transferCheckpointVersion + checkpoint.UpdatedAt = time.Now().UnixMilli() + data, err := json.Marshal(checkpoint) + if err != nil { + return err + } + tempPath := path + ".tmp" + file, err := os.OpenFile(tempPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600) + if err != nil { + return err + } + if _, err = file.Write(data); err == nil { + err = file.Sync() + } + closeErr := file.Close() + if err == nil { + err = closeErr + } + if err != nil { + os.Remove(tempPath) + return err + } + if err := os.Rename(tempPath, path); err != nil { + os.Remove(tempPath) + return err + } + return nil +} + +func transferResponseValidator(headers http.Header) string { + if etag := strings.TrimSpace(headers.Get("ETag")); etag != "" && !strings.HasPrefix(strings.ToUpper(etag), "W/") { + return etag + } + return strings.TrimSpace(headers.Get("Last-Modified")) +} + +func transferTotalLength(resp *http.Response, rangeFrom int64) int64 { + if resp == nil { + return 0 + } + if resp.StatusCode == http.StatusPartialContent { + contentRange := resp.Header.Get("Content-Range") + if slash := strings.LastIndex(contentRange, "/"); slash >= 0 { + if total, err := strconv.ParseInt(contentRange[slash+1:], 10, 64); err == nil { + return total + } + } + if resp.ContentLength > 0 { + return rangeFrom + resp.ContentLength + } + return 0 + } + return resp.ContentLength +} + +func validTransferContentRange(resp *http.Response, rangeFrom int64) bool { + if rangeFrom <= 0 || resp == nil || resp.StatusCode != http.StatusPartialContent { + return true + } + want := fmt.Sprintf("bytes %d-", rangeFrom) + return strings.HasPrefix(resp.Header.Get("Content-Range"), want) +} + +func retryableTransferStatus(status int) bool { + return status == http.StatusRequestTimeout || + status == http.StatusTooEarly || + status == http.StatusTooManyRequests || + status >= 500 +} + +func retryAfterSeconds(resp *http.Response) int { + if resp == nil { + return 0 + } + delay := getRetryAfterDuration(resp) + if delay <= 0 { + return 0 + } + seconds := int(delay.Round(time.Second) / time.Second) + if seconds < 1 { + return 1 + } + return seconds +} + +func transferErrorTypeForStatus(status int, policy DownloadTransferPolicy) string { + if status == http.StatusTooManyRequests { + return "rate_limited" + } + if policy.RefreshStreamOnStatus[status] { + return "expired_stream" + } + if status >= 500 || status == http.StatusRequestTimeout || status == http.StatusTooEarly { + return "transient_network" + } + return "http_error" +} + +func waitTransferRetry(ctx context.Context, delay time.Duration) error { + if delay <= 0 { + return nil + } + return sleepRetry(ctx, delay) +} + +func transferRetryConfig(policy DownloadTransferPolicy) RetryConfig { + return RetryConfig{ + MaxRetries: max(0, policy.MaxAttempts-1), + InitialDelay: policy.InitialRetryDelay, + MaxDelay: policy.MaxRetryDelay, + BackoffFactor: 2, + } +} + +func (r *extensionRuntime) reliableFileDownload( + client *http.Client, + urlStr string, + fullPath string, + headers map[string]string, + onProgress goja.Callable, + trackItemBytes bool, + resumeDownload bool, + persistentCheckpoint bool, + policy DownloadTransferPolicy, +) goja.Value { + unlock := lockDownloadOutputPath(fullPath) + defer unlock() + + callerSetRange := false + for key := range headers { + if strings.EqualFold(key, "Range") { + callerSetRange = true + break + } + } + if callerSetRange { + // A caller-defined range describes a standalone output fragment. It + // cannot safely be combined with a checkpoint owned by this engine. + resumeDownload = false + persistentCheckpoint = false + } + + stagedPath := stagedDownloadPath(fullPath) + checkpointPath := transferCheckpointPath(stagedPath) + fingerprint := transferURLFingerprint(urlStr) + keepPartial := resumeDownload && persistentCheckpoint && fingerprint != "" + checkpoint, checkpointOK := loadTransferCheckpoint(checkpointPath, fingerprint) + if !keepPartial || !checkpointOK { + os.Remove(stagedPath) + os.Remove(checkpointPath) + checkpoint = transferCheckpoint{} + } + + out, err := os.OpenFile(stagedPath, os.O_CREATE|os.O_RDWR, 0600) + if err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to create staged file: %v", err), + }) + } + promoted := false + defer func() { + out.Close() + if promoted { + os.Remove(checkpointPath) + return + } + if !keepPartial { + os.Remove(stagedPath) + os.Remove(checkpointPath) + } + }() + + var written int64 + var validator string + var contentLength int64 + if checkpointOK { + if info, statErr := out.Stat(); statErr == nil { + written = min(checkpoint.Bytes, info.Size()) + validator = checkpoint.Validator + contentLength = checkpoint.Total + if truncateErr := out.Truncate(written); truncateErr != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to restore partial download: %v", truncateErr), + }) + } + } + } + if _, err := out.Seek(written, io.SeekStart); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek staged file: %v", err), + }) + } + + activeItemID := r.getActiveDownloadItemID() + if activeItemID != "" { + SetItemDownloading(activeItemID) + } + shouldTrackItemBytes := activeItemID != "" && trackItemBytes + if shouldTrackItemBytes { + if contentLength > 0 { + SetItemProgress(activeItemID, float64(written)/float64(contentLength), written, contentLength) + } else if written > 0 { + SetItemBytesReceived(activeItemID, written) + } + } + if checkpointOK && written > 0 && contentLength > 0 && written == contentLength { + // The process may have died after the last durable checkpoint but + // before the atomic rename. Publish that already-complete staged file + // without issuing an unsatisfiable Range request at EOF. + if err := out.Sync(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to sync restored download: %v", err), + }) + } + if err := out.Close(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to close restored download: %v", err), + }) + } + if err := os.Rename(stagedPath, fullPath); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to publish restored download: %v", err), + }) + } + promoted = true + os.Remove(checkpointPath) + syncDir(filepath.Dir(fullPath)) + if shouldTrackItemBytes { + SetItemProgress(activeItemID, 1, written, contentLength) + } + return r.jsSuccess(map[string]any{ + "path": fullPath, + "size": written, + "attempts": 0, + "resumed": true, + }) + } + + config := transferRetryConfig(policy) + retryDelay := config.InitialDelay + var lastFailure transferFailure + attemptsUsed := 0 + var lastProgressNotify int64 + lastCheckpointBytes := written + lastCheckpointAt := time.Now() + saveCheckpoint := func() { + if !keepPartial || validator == "" || written <= 0 { + return + } + // Rate-limit checkpoint attempts too: a transient storage failure must + // not turn every subsequent 64 KiB network read into another fsync. + lastCheckpointBytes = written + lastCheckpointAt = time.Now() + // Persist data before the pointer to it. After a power loss, a valid + // checkpoint must never advertise bytes that were only in page cache. + if syncErr := out.Sync(); syncErr != nil { + return + } + _ = saveTransferCheckpoint(checkpointPath, transferCheckpoint{ + Fingerprint: fingerprint, + Validator: validator, + Bytes: written, + Total: contentLength, + }) + } + + for attempt := 1; attempt <= policy.MaxAttempts; attempt++ { + attemptsUsed = attempt + rangeFrom := int64(0) + if resumeDownload && written > 0 && validator != "" { + rangeFrom = written + } + + req, requestErr := http.NewRequest("GET", urlStr, nil) + if requestErr != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "invalid_request", + Message: requestErr.Error(), + Attempts: attempt, + }) + } + req = r.bindDownloadCancelContext(req) + for key, value := range headers { + req.Header.Set(key, value) + } + if req.Header.Get("User-Agent") == "" { + req.Header.Set("User-Agent", appUserAgent()) + } + if rangeFrom > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", rangeFrom)) + req.Header.Set("If-Range", validator) + } + + retryContext := req.Context() + req, watchdog := bindStallWatchdog(req, downloadStallTimeout) + resp, requestErr := client.Do(req) + if requestErr != nil { + stalled := watchdog.stalled.Load() + watchdog.stop() + if activeItemID != "" && isDownloadCancelled(activeItemID) { + return r.jsTransferError(transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }) + } + message := requestErr.Error() + if stalled { + message = fmt.Sprintf( + "download stalled: no data received for %ds (network timeout)", + int(downloadStallTimeout.Seconds()), + ) + } + lastFailure = transferFailure{ + ErrorType: "transient_network", + Message: message, + Attempts: attempt, + } + if attempt == policy.MaxAttempts || retryContext.Err() != nil { + return r.jsTransferError(lastFailure) + } + if written > 0 && (!resumeDownload || validator == "") { + if err := out.Truncate(0); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to restart transfer: %v", err), + Attempts: attempt, + }) + } + written = 0 + if _, err := out.Seek(0, io.SeekStart); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek restarted transfer: %v", err), + Attempts: attempt, + }) + } + } + if err := waitTransferRetry(retryContext, retryDelay); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }) + } + retryDelay = calculateNextDelay(retryDelay, config) + continue + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + retryAfter := retryAfterSeconds(resp) + io.Copy(io.Discard, io.LimitReader(resp.Body, 32*1024)) + resp.Body.Close() + watchdog.stop() + errorType := transferErrorTypeForStatus(resp.StatusCode, policy) + lastFailure = transferFailure{ + ErrorType: errorType, + Message: fmt.Sprintf("HTTP error: %d", resp.StatusCode), + HTTPStatus: resp.StatusCode, + RetryAfterSeconds: retryAfter, + Attempts: attempt, + } + if !retryableTransferStatus(resp.StatusCode) || attempt == policy.MaxAttempts { + return r.jsTransferError(lastFailure) + } + delay := retryDelay + if retryAfter > 0 { + delay = time.Duration(retryAfter) * time.Second + } + if err := waitTransferRetry(retryContext, delay); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }) + } + retryDelay = calculateNextDelay(retryDelay, config) + continue + } + + if rangeFrom > 0 && resp.StatusCode == http.StatusPartialContent && + !validTransferContentRange(resp, rangeFrom) { + contentRange := resp.Header.Get("Content-Range") + resp.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: fmt.Sprintf( + "resume failed: unexpected Content-Range %q at %d bytes", + contentRange, + rangeFrom, + ), + HTTPStatus: resp.StatusCode, + Attempts: attempt, + }) + } + if rangeFrom > 0 && resp.StatusCode == http.StatusPartialContent { + nextValidator := transferResponseValidator(resp.Header) + if nextValidator != "" && validator != "" && nextValidator != validator { + resp.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: "resume failed: response validator changed", + HTTPStatus: resp.StatusCode, + Attempts: attempt, + }) + } + } + if rangeFrom > 0 && resp.StatusCode == http.StatusOK { + if err := out.Truncate(0); err != nil { + resp.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to restart changed transfer: %v", err), + Attempts: attempt, + }) + } + if _, err := out.Seek(0, io.SeekStart); err != nil { + resp.Body.Close() + watchdog.stop() + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek restarted transfer: %v", err), + Attempts: attempt, + }) + } + written = 0 + rangeFrom = 0 + } + + if nextValidator := transferResponseValidator(resp.Header); nextValidator != "" { + validator = nextValidator + } + contentLength = transferTotalLength(resp, rangeFrom) + if callerSetRange { + // The caller asked file.download to materialize only this range; the + // response length, not the complete object's Content-Range total, is + // therefore the integrity boundary for the output file. + contentLength = resp.ContentLength + } + if shouldTrackItemBytes && contentLength > 0 { + SetItemProgress( + activeItemID, + float64(written)/float64(contentLength), + written, + contentLength, + ) + } + var readErr error + buffer := make([]byte, 64*1024) + for { + readCount, bodyErr := resp.Body.Read(buffer) + if readCount > 0 { + watchdog.reset() + writeCount, writeErr := out.Write(buffer[:readCount]) + written += int64(writeCount) + if writeErr != nil || writeCount != readCount { + resp.Body.Close() + watchdog.stop() + if writeErr == nil { + writeErr = io.ErrShortWrite + } + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to write staged file: %v", writeErr), + Attempts: attempt, + }) + } + if shouldTrackItemBytes { + if contentLength > 0 { + SetItemProgress(activeItemID, float64(written)/float64(contentLength), written, contentLength) + } else { + SetItemBytesReceived(activeItemID, written) + } + } + if onProgress != nil && contentLength > 0 && + (written-lastProgressNotify >= progressUpdateThreshold || written >= contentLength) { + lastProgressNotify = written + _, _ = onProgress( + goja.Undefined(), + r.vm.ToValue(written), + r.vm.ToValue(contentLength), + ) + } + if keepPartial && validator != "" && + (written-lastCheckpointBytes >= transferCheckpointBytes || + time.Since(lastCheckpointAt) >= transferCheckpointPeriod) { + saveCheckpoint() + } + } + if bodyErr != nil { + if bodyErr != io.EOF { + readErr = bodyErr + } + break + } + } + resp.Body.Close() + stalled := watchdog.stalled.Load() + watchdog.stop() + + if readErr == nil && contentLength > 0 && written != contentLength { + readErr = io.ErrUnexpectedEOF + } + if readErr == nil { + break + } + + saveCheckpoint() + message := fmt.Sprintf("failed to read response: %v", readErr) + if stalled { + message = fmt.Sprintf( + "download stalled: no data received for %ds (network timeout)", + int(downloadStallTimeout.Seconds()), + ) + } + lastFailure = transferFailure{ + ErrorType: "transient_network", + Message: message, + Attempts: attempt, + } + if attempt == policy.MaxAttempts || + (activeItemID != "" && isDownloadCancelled(activeItemID)) { + if activeItemID != "" && isDownloadCancelled(activeItemID) { + lastFailure.ErrorType = "cancelled" + lastFailure.Message = "download cancelled" + } + return r.jsTransferError(lastFailure) + } + + if !resumeDownload || validator == "" { + if err := out.Truncate(0); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to restart transfer: %v", err), + Attempts: attempt, + }) + } + if _, err := out.Seek(0, io.SeekStart); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to seek restarted transfer: %v", err), + Attempts: attempt, + }) + } + written = 0 + validator = "" + os.Remove(checkpointPath) + } + if err := waitTransferRetry(retryContext, retryDelay); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "cancelled", + Message: "download cancelled", + Attempts: attempt, + }) + } + retryDelay = calculateNextDelay(retryDelay, config) + } + + if written <= 0 { + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: "download response was empty", + Attempts: attemptsUsed, + }) + } + if contentLength > 0 && written != contentLength { + return r.jsTransferError(transferFailure{ + ErrorType: "integrity_failed", + Message: fmt.Sprintf( + "download size mismatch: expected %d bytes, wrote %d", + contentLength, + written, + ), + Attempts: attemptsUsed, + }) + } + if err := out.Sync(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to sync staged file: %v", err), + }) + } + if err := out.Close(); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to finalize staged file: %v", err), + }) + } + if err := os.Rename(stagedPath, fullPath); err != nil { + return r.jsTransferError(transferFailure{ + ErrorType: "storage_error", + Message: fmt.Sprintf("failed to publish file: %v", err), + }) + } + promoted = true + os.Remove(checkpointPath) + syncDir(filepath.Dir(fullPath)) + if shouldTrackItemBytes { + if contentLength > 0 { + SetItemProgress(activeItemID, 1, written, contentLength) + } else if written > 0 { + SetItemBytesReceived(activeItemID, written) + } + } + + GoLog( + "[Extension:%s] Reliable transfer downloaded %d bytes to %s\n", + r.extensionID, + written, + fullPath, + ) + return r.jsSuccess(map[string]any{ + "path": fullPath, + "size": written, + "attempts": attemptsUsed, + }) +} diff --git a/go_backend/extension_transfer_policy.go b/go_backend/extension_transfer_policy.go new file mode 100644 index 00000000..f6276194 --- /dev/null +++ b/go_backend/extension_transfer_policy.go @@ -0,0 +1,215 @@ +package gobackend + +import ( + "fmt" + "math" + "strings" + "time" +) + +const ( + defaultTransferMaxAttempts = 3 + defaultTransferInitialRetryDelay = 500 * time.Millisecond + defaultTransferMaxRetryDelay = 8 * time.Second + defaultParallelSegments = 3 + maxParallelSegments = 8 + maxExtensionDownloadConcurrency = 3 +) + +// DownloadTransferPolicy is the generic manifest contract used by every +// extension-backed transfer. It deliberately describes transport behavior, +// never a provider name, so new extensions can opt into the same reliability +// and concurrency features without changes in the app. +// +// Manifests declare it under capabilities.downloadTransfer: +// +// { +// "maxAttempts": 4, +// "initialRetryDelayMs": 500, +// "maxRetryDelayMs": 8000, +// "resumePolicy": "validated", +// "persistentCheckpoint": true, +// "refreshStreamOnStatus": [401, 403], +// "maxParallelSegments": 4, +// "maxConcurrentDownloads": 2 +// } +type DownloadTransferPolicy struct { + MaxAttempts int + InitialRetryDelay time.Duration + MaxRetryDelay time.Duration + ResumePolicy string + PersistentCheckpoint bool + RefreshStreamOnStatus map[int]bool + MaxParallelSegments int + MaxConcurrentDownloads int +} + +func defaultDownloadTransferPolicy() DownloadTransferPolicy { + return DownloadTransferPolicy{ + MaxAttempts: defaultTransferMaxAttempts, + InitialRetryDelay: defaultTransferInitialRetryDelay, + MaxRetryDelay: defaultTransferMaxRetryDelay, + ResumePolicy: "none", + PersistentCheckpoint: false, + RefreshStreamOnStatus: map[int]bool{httpStatusUnauthorized: true, httpStatusForbidden: true}, + MaxParallelSegments: defaultParallelSegments, + MaxConcurrentDownloads: maxExtensionDownloadConcurrency, + } +} + +const ( + httpStatusUnauthorized = 401 + httpStatusForbidden = 403 +) + +func capabilityObject(capabilities map[string]any, key string) map[string]any { + if capabilities == nil { + return nil + } + value, ok := capabilities[key] + if !ok { + return nil + } + switch typed := value.(type) { + case map[string]any: + return typed + default: + return nil + } +} + +func capabilityInt(value any, fallback int) int { + switch typed := value.(type) { + case int: + return typed + case int32: + return int(typed) + case int64: + return int(typed) + case float32: + return int(math.Round(float64(typed))) + case float64: + return int(math.Round(typed)) + default: + return fallback + } +} + +func clampInt(value, minimum, maximum int) int { + if value < minimum { + return minimum + } + if value > maximum { + return maximum + } + return value +} + +func parseRefreshStatuses(value any, fallback map[int]bool) map[int]bool { + values, ok := value.([]any) + if !ok { + return fallback + } + parsed := make(map[int]bool) + for _, raw := range values { + status := capabilityInt(raw, 0) + if status >= 400 && status <= 599 { + parsed[status] = true + } + } + if len(parsed) == 0 { + return fallback + } + return parsed +} + +func (m *ExtensionManifest) DownloadTransferPolicy() DownloadTransferPolicy { + policy := defaultDownloadTransferPolicy() + if m == nil { + return policy + } + config := capabilityObject(m.Capabilities, "downloadTransfer") + if config == nil { + return policy + } + + policy.MaxAttempts = clampInt( + capabilityInt(config["maxAttempts"], policy.MaxAttempts), + 1, + 8, + ) + initialDelayMs := clampInt( + capabilityInt(config["initialRetryDelayMs"], int(policy.InitialRetryDelay/time.Millisecond)), + 100, + 30_000, + ) + maxDelayMs := clampInt( + capabilityInt(config["maxRetryDelayMs"], int(policy.MaxRetryDelay/time.Millisecond)), + initialDelayMs, + 120_000, + ) + policy.InitialRetryDelay = time.Duration(initialDelayMs) * time.Millisecond + policy.MaxRetryDelay = time.Duration(maxDelayMs) * time.Millisecond + + if value, ok := config["resumePolicy"].(string); ok { + switch strings.ToLower(strings.TrimSpace(value)) { + case "validated", "none": + policy.ResumePolicy = strings.ToLower(strings.TrimSpace(value)) + } + } + if value, ok := config["persistentCheckpoint"].(bool); ok { + policy.PersistentCheckpoint = value && policy.ResumePolicy == "validated" + } + policy.RefreshStreamOnStatus = parseRefreshStatuses( + config["refreshStreamOnStatus"], + policy.RefreshStreamOnStatus, + ) + policy.MaxParallelSegments = clampInt( + capabilityInt(config["maxParallelSegments"], policy.MaxParallelSegments), + 1, + maxParallelSegments, + ) + policy.MaxConcurrentDownloads = clampInt( + capabilityInt(config["maxConcurrentDownloads"], policy.MaxConcurrentDownloads), + 1, + maxExtensionDownloadConcurrency, + ) + return policy +} + +func validateDownloadTransferCapability(capabilities map[string]any) error { + if capabilities == nil { + return nil + } + _, exists := capabilities["downloadTransfer"] + if !exists { + return nil + } + config := capabilityObject(capabilities, "downloadTransfer") + if config == nil { + return fmt.Errorf("must be an object") + } + if rawResume, ok := config["resumePolicy"]; ok { + resume, ok := rawResume.(string) + if !ok || (resume != "none" && resume != "validated") { + return fmt.Errorf("resumePolicy must be 'none' or 'validated'") + } + } + if rawCheckpoint, ok := config["persistentCheckpoint"]; ok { + if _, ok := rawCheckpoint.(bool); !ok { + return fmt.Errorf("persistentCheckpoint must be a boolean") + } + } + for _, key := range []string{ + "maxAttempts", + "initialRetryDelayMs", + "maxRetryDelayMs", + "maxParallelSegments", + "maxConcurrentDownloads", + } { + if rawValue, ok := config[key]; ok && capabilityInt(rawValue, -1) < 0 { + return fmt.Errorf("%s must be a non-negative number", key) + } + } + return nil +} diff --git a/go_backend/extension_transfer_policy_test.go b/go_backend/extension_transfer_policy_test.go new file mode 100644 index 00000000..8e645722 --- /dev/null +++ b/go_backend/extension_transfer_policy_test.go @@ -0,0 +1,51 @@ +package gobackend + +import ( + "testing" + "time" +) + +func TestDownloadTransferPolicyParsesAndBoundsManifestCapability(t *testing.T) { + manifest := &ExtensionManifest{Capabilities: map[string]any{ + "downloadTransfer": map[string]any{ + "maxAttempts": float64(20), + "initialRetryDelayMs": float64(25), + "maxRetryDelayMs": float64(50), + "resumePolicy": "validated", + "persistentCheckpoint": true, + "refreshStreamOnStatus": []any{float64(401), float64(410)}, + "maxParallelSegments": float64(99), + "maxConcurrentDownloads": float64(99), + }, + }} + + policy := manifest.DownloadTransferPolicy() + if policy.MaxAttempts != 8 || policy.InitialRetryDelay != 100*time.Millisecond || + policy.MaxRetryDelay != 100*time.Millisecond || policy.ResumePolicy != "validated" || + !policy.PersistentCheckpoint || !policy.RefreshStreamOnStatus[401] || + !policy.RefreshStreamOnStatus[410] || policy.MaxParallelSegments != 8 || + policy.MaxConcurrentDownloads != 3 { + t.Fatalf("unexpected policy: %#v", policy) + } +} + +func TestDownloadTransferCapabilityValidation(t *testing.T) { + valid := map[string]any{"downloadTransfer": map[string]any{ + "resumePolicy": "validated", + "persistentCheckpoint": true, + }} + if err := validateDownloadTransferCapability(valid); err != nil { + t.Fatalf("valid capability rejected: %v", err) + } + + invalid := []map[string]any{ + {"downloadTransfer": "yes"}, + {"downloadTransfer": map[string]any{"resumePolicy": "unsafe"}}, + {"downloadTransfer": map[string]any{"persistentCheckpoint": "yes"}}, + } + for _, capabilities := range invalid { + if err := validateDownloadTransferCapability(capabilities); err == nil { + t.Fatalf("invalid capability accepted: %#v", capabilities) + } + } +}