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

@@ -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

.vscode/launch.json

@@ -18,6 +18,9 @@
             ],
             "envFile": "${workspaceFolder}/machinery/.env.local",
             "buildFlags": "--tags dynamic",
+            "env": {
+                "GOWORK": "off"
+            },
         },
         {
             "name": "Launch React",

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=

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

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.
+		// (4) Stream the remaining bytes to the vault via PATCH, reading directly
-		if _, sErr := file.Seek(offset, io.SeekStart); sErr != nil {
+		// from disk so the recording is never fully buffered in memory. When a chunk
-			return false, false, true, "", sErr
+		// size is configured the data is sent across several PATCH requests,
-		}
+		// checkpointing the offset after each one so an interruption resumes from the
-		newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, size, file, vault, publicKey, deviceKey)
+		// last completed chunk instead of re-uploading everything.
-		if perr != nil {
+		chunkSize := tusChunkSize()
-			if status >= 400 {
+		progressed := false
-				// Definitive rejection (e.g. provider push failed during finalize).
+		patchFailed := false
-				// Re-evaluate via HEAD on the next iteration to decide retry/restart.
+		var lastBody string
-				log.Log.Info(label + ": resumable patch rejected, " + perr.Error())
+		for offset < size {
-			} else {
+			// Re-seek every chunk so the on-disk position always matches the
-				log.Log.Info(label + ": resumable patch failed, " + perr.Error())
+			// 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.
+		// All declared bytes have been sent and acknowledged: the upload is done.
-		saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
+		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) {
+func tusPatch(client *http.Client, uploadURL string, offset, length 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, length))
-	req, err := http.NewRequest("PATCH", uploadURL, io.LimitReader(file, remaining))
 	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.
+		// A 204 without a parseable offset means this PATCH was fully accepted.
-		return size, resp.StatusCode, respBody, nil
+		return offset + length, resp.StatusCode, respBody, nil
 	}
 	return newOffset, resp.StatusCode, respBody, nil
 }

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

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
 		}

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)
 	}