From 5973ba025d933136a66bc5b43d0f707f9592080e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Verstraeten?= Date: Thu, 11 Jun 2026 23:03:05 +0200 Subject: [PATCH] Add configurable tus chunking and stability fixes Enable configurable chunked tus uploads and related robustness changes. - Add AGENT_TUS_CHUNK_SIZE_BYTES env (default 1 MiB, 0 disables chunking) and docs in machinery/.env; ignore machinery/go.work files and set GOWORK=off in VSCode launch to avoid go.work during debugging. - Implement tusChunkSize() and update uploadVaultResumable to send PATCHes in configurable chunk sizes, checkpoint progress after each chunk, and handle partial failures by refreshing retry budget when progress occurs. - Change tusPatch to accept an explicit length and return the advanced offset when a PATCH is fully accepted. - Skip uploads when the recording file no longer exists to avoid infinite retries. - Add tests exercising chunked uploads, chunking-disabled behavior, and tusChunkSize parsing; extend fakeTus test server to record patch sizes. - Reduce noisy info logs to debug in AAC transcoder and WebRTC audio processing. These changes improve resumable upload reliability, allow tuning for proxy/load-balancer limits, and reduce log spam during normal operation. --- .gitignore | 2 + .vscode/launch.json | 3 + machinery/.env | 7 ++ machinery/src/cloud/kerberos_vault.go | 9 ++ machinery/src/cloud/tus_client.go | 108 +++++++++++++++----- machinery/src/cloud/tus_client_test.go | 101 ++++++++++++++++++ machinery/src/webrtc/aac_transcoder_stub.go | 8 +- machinery/src/webrtc/main.go | 4 +- 8 files changed, 210 insertions(+), 32 deletions(-) diff --git a/.gitignore b/.gitignore index ea402d3..da6c86f 100644 --- a/.gitignore +++ b/.gitignore @@ -14,5 +14,7 @@ machinery/test* machinery/init-dev.sh machinery/.env.local machinery/vendor +machinery/go.work +machinery/go.work.sum deployments/docker/private-docker-compose.yaml video.mp4 \ No newline at end of file diff --git a/.vscode/launch.json b/.vscode/launch.json index 219c74d..d552cd7 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -18,6 +18,9 @@ ], "envFile": "${workspaceFolder}/machinery/.env.local", "buildFlags": "--tags dynamic", + "env": { + "GOWORK": "off" + }, }, { "name": "Launch React", diff --git a/machinery/.env b/machinery/.env index 16ba7d3..23c1752 100644 --- a/machinery/.env +++ b/machinery/.env @@ -27,5 +27,12 @@ AGENT_KERBEROSVAULT_SECONDARY_DIRECTORY= AGENT_KERBEROSVAULT_SECONDARY_ACCESS_KEY= AGENT_KERBEROSVAULT_SECONDARY_SECRET_KEY= +# Resumable (tus) uploads to Kerberos Vault are enabled by default. +# Set to true to fall back to the legacy single-shot POST /storage upload. +#AGENT_DISABLE_RESUMABLE_UPLOAD=false +# Bytes sent per PATCH request (default 1 MiB = 1048576). 0 disables chunking +# and sends the whole file in a single PATCH. +AGENT_TUS_CHUNK_SIZE_BYTES=1048576 + # Open telemetry tracing endpoint OTEL_EXPORTER_OTLP_ENDPOINT= \ No newline at end of file diff --git a/machinery/src/cloud/kerberos_vault.go b/machinery/src/cloud/kerberos_vault.go index 76070fe..1253d73 100644 --- a/machinery/src/cloud/kerberos_vault.go +++ b/machinery/src/cloud/kerberos_vault.go @@ -30,6 +30,15 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) ( return false, false, errors.New(err) } + // If the recording no longer exists on disk there is nothing to upload. + // This can happen when the file was already removed (e.g. cleanup, or an + // earlier successful upload). Skip it so the watcher drops the marker + // instead of retrying indefinitely. + if _, err := os.Stat("data/recordings/" + fileName); err != nil { + log.Log.Info("UploadKerberosVault: skipping " + fileName + ", file doesn't exist anymore") + return false, false, nil + } + // timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token // 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4 // - Timestamp diff --git a/machinery/src/cloud/tus_client.go b/machinery/src/cloud/tus_client.go index d88dcfb..f4fc9b8 100644 --- a/machinery/src/cloud/tus_client.go +++ b/machinery/src/cloud/tus_client.go @@ -42,6 +42,31 @@ func resumableUploadsEnabled() bool { return os.Getenv("AGENT_DISABLE_RESUMABLE_UPLOAD") != "true" } +// tusDefaultChunkSize is the number of bytes uploaded per PATCH request when no +// explicit size is configured. Splitting the upload into chunks keeps each HTTP +// request small enough for intermediary proxies/load balancers and checkpoints +// progress frequently, so an interruption resumes with minimal re-upload. +const tusDefaultChunkSize int64 = 1 << 20 // 1 MiB + +// tusChunkSize returns the number of bytes to send per PATCH request. It +// defaults to tusDefaultChunkSize (1 MiB) and can be overridden with the +// AGENT_TUS_CHUNK_SIZE_BYTES environment variable. A value of 0 (or negative) +// disables chunking and sends the remaining bytes in a single PATCH. +func tusChunkSize() int64 { + v := os.Getenv("AGENT_TUS_CHUNK_SIZE_BYTES") + if v == "" { + return tusDefaultChunkSize + } + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + return tusDefaultChunkSize + } + if n <= 0 { + return 0 // chunking disabled: send everything in one PATCH + } + return n +} + // uploadVaultResumable uploads a recording to a Kerberos Vault using the tus // resumable upload protocol. // @@ -136,29 +161,61 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, continue } - // (4) Stream the remaining bytes in a single PATCH directly from disk. - if _, sErr := file.Seek(offset, io.SeekStart); sErr != nil { - return false, false, true, "", sErr - } - newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, size, file, vault, publicKey, deviceKey) - if perr != nil { - if status >= 400 { - // Definitive rejection (e.g. provider push failed during finalize). - // Re-evaluate via HEAD on the next iteration to decide retry/restart. - log.Log.Info(label + ": resumable patch rejected, " + perr.Error()) - } else { - log.Log.Info(label + ": resumable patch failed, " + perr.Error()) + // (4) Stream the remaining bytes to the vault via PATCH, reading directly + // from disk so the recording is never fully buffered in memory. When a chunk + // size is configured the data is sent across several PATCH requests, + // checkpointing the offset after each one so an interruption resumes from the + // last completed chunk instead of re-uploading everything. + chunkSize := tusChunkSize() + progressed := false + patchFailed := false + var lastBody string + for offset < size { + // Re-seek every chunk so the on-disk position always matches the + // server-acknowledged offset, even if a PATCH was partially accepted. + if _, sErr := file.Seek(offset, io.SeekStart); sErr != nil { + return false, false, true, "", sErr + } + patchLen := size - offset + if chunkSize > 0 && chunkSize < patchLen { + patchLen = chunkSize + } + newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, patchLen, file, vault, publicKey, deviceKey) + if perr != nil { + if status >= 400 { + // Definitive rejection (e.g. provider push failed during finalize). + // Re-evaluate via HEAD on the next iteration to decide retry/restart. + log.Log.Info(label + ": resumable patch rejected, " + perr.Error()) + } else { + log.Log.Info(label + ": resumable patch failed, " + perr.Error()) + } + tusBackoff(attempt) + patchFailed = true + break + } + if newOffset > offset { + progressed = true + } + offset = newOffset + lastBody = respBody + if offset < size { + // Partial progress: persist so a later retry resumes from here. + saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size}) + } + } + if patchFailed { + if progressed { + // Forward progress refreshes the retry budget: maxAttempts bounds the + // number of consecutive failures, not the number of chunks needed for + // a large recording. + attempt = -1 } - tusBackoff(attempt) continue } - if newOffset >= size { - removeTusResumeState(sidecar) - return true, true, true, respBody, nil - } - // Partial progress: persist and continue with the next chunk. - saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size}) + // All declared bytes have been sent and acknowledged: the upload is done. + removeTusResumeState(sidecar) + return true, true, true, lastBody, nil } return false, true, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries") @@ -227,16 +284,15 @@ func tusHead(client *http.Client, uploadURL string, vault models.KStorage, publi return offset, resp.StatusCode, nil } -// tusPatch streams the remaining bytes of the file (from offset to size) to the +// tusPatch streams up to length bytes of the file (starting at offset) to the // upload URL using a single PATCH request. The body is read straight from the // *os.File, so the recording is never fully buffered in memory. -func tusPatch(client *http.Client, uploadURL string, offset, size int64, file io.Reader, vault models.KStorage, publicKey, deviceKey string) (int64, int, string, error) { - remaining := size - offset - req, err := http.NewRequest("PATCH", uploadURL, io.LimitReader(file, remaining)) +func tusPatch(client *http.Client, uploadURL string, offset, length int64, file io.Reader, vault models.KStorage, publicKey, deviceKey string) (int64, int, string, error) { + req, err := http.NewRequest("PATCH", uploadURL, io.LimitReader(file, length)) if err != nil { return offset, 0, "", err } - req.ContentLength = remaining + req.ContentLength = length req.Header.Set("Tus-Resumable", tusResumableVersion) req.Header.Set("Content-Type", "application/offset+octet-stream") req.Header.Set("Upload-Offset", strconv.FormatInt(offset, 10)) @@ -258,8 +314,8 @@ func tusPatch(client *http.Client, uploadURL string, offset, size int64, file io newOffsetStr := resp.Header.Get("Upload-Offset") newOffset, perr := strconv.ParseInt(newOffsetStr, 10, 64) if perr != nil { - // A 204 without a parseable offset means the upload finished. - return size, resp.StatusCode, respBody, nil + // A 204 without a parseable offset means this PATCH was fully accepted. + return offset + length, resp.StatusCode, respBody, nil } return newOffset, resp.StatusCode, respBody, nil } diff --git a/machinery/src/cloud/tus_client_test.go b/machinery/src/cloud/tus_client_test.go index 3736ab4..3e61ab3 100644 --- a/machinery/src/cloud/tus_client_test.go +++ b/machinery/src/cloud/tus_client_test.go @@ -30,6 +30,7 @@ type fakeTus struct { counter int creates int lastPatchBytes int64 + patchSizes []int64 // unsupported makes the creation endpoint return 404, simulating an older // vault without a tus endpoint. @@ -68,6 +69,15 @@ func (s *fakeTus) lastPatch() int64 { return s.lastPatchBytes } +// patchCounts returns the number of PATCH requests received and the size of each. +func (s *fakeTus) patchCounts() (int, []int64) { + s.mu.Lock() + defer s.mu.Unlock() + sizes := make([]int64, len(s.patchSizes)) + copy(sizes, s.patchSizes) + return len(s.patchSizes), sizes +} + func (s *fakeTus) createCount() int { s.mu.Lock() defer s.mu.Unlock() @@ -118,6 +128,7 @@ func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.mu.Lock() u.offset += n s.lastPatchBytes = n + s.patchSizes = append(s.patchSizes, n) complete := u.offset >= u.size failNow := complete && s.failFinalize > 0 if failNow { @@ -202,6 +213,96 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) { } } +func TestUploadVaultResumable_Chunked(t *testing.T) { + srv := newFakeTus() + ts := httptest.NewServer(srv) + defer ts.Close() + + fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4" + // 10 KiB payload uploaded in 4 KiB chunks => 3 PATCH requests (4096+4096+2048). + payload := bytes.Repeat([]byte("c"), 10240) + withRecording(t, fileName, payload) + t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "4096") + + uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !uploaded || !supported { + t.Fatalf("expected chunked upload success, got uploaded=%v supported=%v", uploaded, supported) + } + if got := srv.totalBytes(); got != int64(len(payload)) { + t.Fatalf("server received %d bytes, want %d", got, len(payload)) + } + count, sizes := srv.patchCounts() + if count != 3 { + t.Fatalf("expected 3 chunked PATCH requests, got %d (sizes=%v)", count, sizes) + } + want := []int64{4096, 4096, 2048} + for i, w := range want { + if sizes[i] != w { + t.Fatalf("chunk %d size = %d, want %d (sizes=%v)", i, sizes[i], w, sizes) + } + } + if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) { + t.Fatalf("expected sidecar removed after success, stat err = %v", err) + } +} + +func TestUploadVaultResumable_ChunkingDisabled(t *testing.T) { + srv := newFakeTus() + ts := httptest.NewServer(srv) + defer ts.Close() + + fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4" + payload := bytes.Repeat([]byte("d"), 10240) + withRecording(t, fileName, payload) + // 0 disables chunking: the whole file should go out in a single PATCH. + t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "0") + + uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !uploaded || !supported { + t.Fatalf("expected success, got uploaded=%v supported=%v", uploaded, supported) + } + count, sizes := srv.patchCounts() + if count != 1 { + t.Fatalf("expected a single PATCH when chunking is disabled, got %d (sizes=%v)", count, sizes) + } + if sizes[0] != int64(len(payload)) { + t.Fatalf("single PATCH size = %d, want %d", sizes[0], len(payload)) + } +} + +func TestTusChunkSize(t *testing.T) { + cases := []struct { + name string + env string + set bool + want int64 + }{ + {name: "default when unset", set: false, want: tusDefaultChunkSize}, + {name: "default on invalid", env: "notanumber", set: true, want: tusDefaultChunkSize}, + {name: "explicit value", env: "65536", set: true, want: 65536}, + {name: "zero disables", env: "0", set: true, want: 0}, + {name: "negative disables", env: "-5", set: true, want: 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if tc.set { + t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", tc.env) + } else { + t.Setenv("AGENT_TUS_CHUNK_SIZE_BYTES", "") + } + if got := tusChunkSize(); got != tc.want { + t.Fatalf("tusChunkSize() = %d, want %d", got, tc.want) + } + }) + } +} + func TestUploadVaultResumable_Unsupported(t *testing.T) { srv := newFakeTus() srv.unsupported = true diff --git a/machinery/src/webrtc/aac_transcoder_stub.go b/machinery/src/webrtc/aac_transcoder_stub.go index 1b0a8cf..e326ed0 100644 --- a/machinery/src/webrtc/aac_transcoder_stub.go +++ b/machinery/src/webrtc/aac_transcoder_stub.go @@ -96,7 +96,7 @@ func NewAACTranscoder() (*AACTranscoder, error) { buffered := t.outBuf.Len() t.outMu.Unlock() if buffered <= 8192 || buffered%16000 == 0 { - log.Log.Info("webrtc.aac_transcoder: ffmpeg produced PCMU bytes, buffered=" + strconv.Itoa(buffered)) + log.Log.Debug("webrtc.aac_transcoder: ffmpeg produced PCMU bytes, buffered=" + strconv.Itoa(buffered)) } } if readErr != nil { @@ -129,14 +129,14 @@ func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) { return nil, err } if len(adtsData) <= 512 || len(adtsData)%1024 == 0 { - log.Log.Info("webrtc.aac_transcoder: wrote AAC bytes to ffmpeg, input=" + strconv.Itoa(len(adtsData))) + log.Log.Debug("webrtc.aac_transcoder: wrote AAC bytes to ffmpeg, input=" + strconv.Itoa(len(adtsData))) } deadline := time.Now().Add(75 * time.Millisecond) for { data := t.readAvailable() if len(data) > 0 { - log.Log.Info("webrtc.aac_transcoder: returning PCMU bytes=" + strconv.Itoa(len(data))) + log.Log.Debug("webrtc.aac_transcoder: returning PCMU bytes=" + strconv.Itoa(len(data))) return data, nil } @@ -144,7 +144,7 @@ func (t *AACTranscoder) Transcode(adtsData []byte) ([]byte, error) { if stderr := t.stderrString(); stderr != "" { log.Log.Warning("webrtc.aac_transcoder: no output before deadline, ffmpeg stderr: " + stderr) } else { - log.Log.Info("webrtc.aac_transcoder: no PCMU output before deadline") + log.Log.Debug("webrtc.aac_transcoder: no PCMU output before deadline") } return nil, nil } diff --git a/machinery/src/webrtc/main.go b/machinery/src/webrtc/main.go index b73641d..a6d1c48 100644 --- a/machinery/src/webrtc/main.go +++ b/machinery/src/webrtc/main.go @@ -988,7 +988,7 @@ func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster if len(pcmu) == 0 { state.aacNoOutput++ if state.aacNoOutput <= 5 || state.aacNoOutput%100 == 0 { - log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet produced no PCMU output yet (aac_packets=%d, no_output=%d, input_bytes=%d)", state.aacPacketsSeen, state.aacNoOutput, len(pkt.Data))) + log.Log.Debug(fmt.Sprintf("webrtc.main.processAudioPacket(): AAC packet produced no PCMU output yet (aac_packets=%d, no_output=%d, input_bytes=%d)", state.aacPacketsSeen, state.aacNoOutput, len(pkt.Data))) } return // decoder still buffering } @@ -1004,7 +1004,7 @@ func processAudioPacket(pkt packets.Packet, state *streamState, audioBroadcaster state.lastAudioSample.Duration = sampleDuration(pkt, state.lastAudioSample.PacketTimestamp, 20*time.Millisecond) state.audioSamplesSent++ if state.audioSamplesSent <= 5 || state.audioSamplesSent%100 == 0 { - log.Log.Info(fmt.Sprintf("webrtc.main.processAudioPacket(): queueing audio sample (samples=%d, codec=%s, bytes=%d, duration_ms=%d, peers=%d)", state.audioSamplesSent, pkt.Codec, len(state.lastAudioSample.Data), state.lastAudioSample.Duration.Milliseconds(), audioBroadcaster.PeerCount())) + log.Log.Debug(fmt.Sprintf("webrtc.main.processAudioPacket(): queueing audio sample (samples=%d, codec=%s, bytes=%d, duration_ms=%d, peers=%d)", state.audioSamplesSent, pkt.Codec, len(state.lastAudioSample.Data), state.lastAudioSample.Duration.Milliseconds(), audioBroadcaster.PeerCount())) } audioBroadcaster.WriteSample(*state.lastAudioSample) }