Merge pull request #286 from kerberos-io/feature/resumable-uploads-tusd

feature/resumable-uploads-tusd

.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=true
+# 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/cmd/mp4analyze/main.go

@@ -0,0 +1,425 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/Eyevinn/mp4ff/avc"
+	mp4ff "github.com/Eyevinn/mp4ff/mp4"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		fmt.Println("usage: mp4analyze <file.mp4>")
+		os.Exit(1)
+	}
+	f, err := os.Open(os.Args[1])
+	if err != nil {
+		panic(err)
+	}
+	defer f.Close()
+	parsed, err := mp4ff.DecodeFile(f)
+	if err != nil {
+		panic(err)
+	}
+
+	// Movie-level info
+	if parsed.Init != nil && parsed.Init.Moov != nil {
+		moov := parsed.Init.Moov
+		fmt.Printf("ftyp/moov present. timescale(mvhd)=%d duration(mvhd)=%d\n",
+			moov.Mvhd.Timescale, moov.Mvhd.Duration)
+		for _, trak := range moov.Traks {
+			ts := trak.Mdia.Mdhd.Timescale
+			fmt.Printf("  trak id=%d handler=%s mdhd.timescale=%d mdhd.duration=%d\n",
+				trak.Tkhd.TrackID, trak.Mdia.Hdlr.HandlerType, ts, trak.Mdia.Mdhd.Duration)
+		}
+	} else {
+		fmt.Println("no Init/Moov (pure fragmented stream?)")
+	}
+
+	// sidx vs actual segment layout. MSE players use sidx to map presentation
+	// time -> byte ranges; if sidx references disagree with the real segment
+	// sizes/durations (e.g. after an early/short flush) the player fetches the
+	// wrong bytes and fails to decode — a failure that "heals" on seek.
+	fmt.Println("=== sidx references vs actual segments ===")
+	var sidxRefs []mp4ff.SidxRef
+	for _, c := range parsed.Children {
+		if s, ok := c.(*mp4ff.SidxBox); ok {
+			fmt.Printf("  sidx: timescale=%d earliestPresTime=%d firstOffset=%d refCount=%d anchor(after sidx)=%d\n",
+				s.Timescale, s.EarliestPresentationTime, s.FirstOffset, len(s.SidxRefs), s.AnchorPoint)
+			sidxRefs = s.SidxRefs
+		}
+	}
+	// Actual segment sizes (styp+moof+mdat) and fragment durations.
+	type segInfo struct {
+		size uint64
+		dur  uint64
+	}
+	var actual []segInfo
+	for _, seg := range parsed.Segments {
+		var sz uint64
+		if seg.Styp != nil {
+			sz += seg.Styp.Size()
+		}
+		if seg.Sidx != nil {
+			sz += seg.Sidx.Size()
+		}
+		var dur uint64
+		for _, fr := range seg.Fragments {
+			sz += fr.Moof.Size()
+			if fr.Mdat != nil {
+				sz += fr.Mdat.Size()
+			}
+			for _, traf := range fr.Moof.Trafs {
+				if traf.Tfhd.TrackID != 1 {
+					continue
+				}
+				for _, trun := range traf.Truns {
+					for _, s := range trun.Samples {
+						dur += uint64(s.Dur)
+					}
+				}
+			}
+		}
+		actual = append(actual, segInfo{size: sz, dur: dur})
+	}
+	for i := range actual {
+		refStr := "(no sidx ref)"
+		if i < len(sidxRefs) {
+			r := sidxRefs[i]
+			mark := ""
+			if uint64(r.ReferencedSize) != actual[i].size {
+				mark += fmt.Sprintf(" SIZE MISMATCH actual=%d", actual[i].size)
+			}
+			if uint64(r.SubSegmentDuration) != actual[i].dur {
+				mark += fmt.Sprintf(" DUR MISMATCH actual=%d", actual[i].dur)
+			}
+			refStr = fmt.Sprintf("sidx.size=%d sidx.dur=%d type=%d sap=%d/%d%s",
+				r.ReferencedSize, r.SubSegmentDuration, r.ReferenceType, r.StartsWithSAP, r.SAPType, mark)
+		}
+		fmt.Printf("  seg%02d actual.size=%d actual.dur=%d | %s\n", i, actual[i].size, actual[i].dur, refStr)
+	}
+
+	fmt.Println("=== fragments ===")
+	fragIdx := 0
+	var allKeyGlobal []uint64 // global keyframe decode times (track timescale units)
+	var prevTfdtEnd = map[uint32]uint64{}
+	for si, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			for _, traf := range fr.Moof.Trafs {
+				tid := traf.Tfhd.TrackID
+				tfdt := traf.Tfdt.BaseMediaDecodeTime()
+				offset := uint64(0)
+				var keys []uint64    // keyframe offset-from-tfdt
+				var durs []uint64
+				zeroDur := 0
+				nSamples := 0
+				for _, trun := range traf.Truns {
+					for _, s := range trun.Samples {
+						nSamples++
+						if (s.Flags>>24)&0x03 == 0x02 { // sample_depends_on==2 => IDR/sync
+							keys = append(keys, offset)
+							if tid == 1 {
+								allKeyGlobal = append(allKeyGlobal, tfdt+offset)
+							}
+						}
+						if s.Dur == 0 {
+							zeroDur++
+						}
+						durs = append(durs, uint64(s.Dur))
+						offset += uint64(s.Dur)
+					}
+				}
+				cont := ""
+				if pe, ok := prevTfdtEnd[tid]; ok {
+					if tfdt != pe {
+						cont = fmt.Sprintf(" <-- tfdt GAP/JUMP prev_end=%d delta=%d", pe, int64(tfdt)-int64(pe))
+					}
+				}
+				prevTfdtEnd[tid] = tfdt + offset
+				if tid == 1 {
+					// in-fragment keyframe gaps
+					var gaps []int64
+					for i := 1; i < len(keys); i++ {
+						gaps = append(gaps, int64(keys[i])-int64(keys[i-1]))
+					}
+					fmt.Printf("seg%d frag%d trk%d tfdt=%d dur=%d nSamp=%d zeroDur=%d keys=%v inFragKeyGaps=%v%s\n",
+						si, fragIdx, tid, tfdt, offset, nSamples, zeroDur, keys, gaps, cont)
+				}
+			}
+			fragIdx++
+		}
+	}
+
+	fmt.Println("=== global video keyframe decode times & gaps ===")
+	for i, k := range allKeyGlobal {
+		gap := int64(0)
+		if i > 0 {
+			gap = int64(k) - int64(allKeyGlobal[i-1])
+		}
+		flag := ""
+		if i > 1 {
+			prevGap := int64(allKeyGlobal[i-1]) - int64(allKeyGlobal[i-2])
+			if gap > 0 && prevGap > 0 && gap*2 < prevGap {
+				flag = fmt.Sprintf("  <== SEAM? gap=%d < prevGap/2=%d", gap, prevGap/2)
+			}
+		}
+		fmt.Printf("  kf#%02d dt=%d gap=%d%s\n", i, k, gap, flag)
+	}
+
+	// Full sample timeline: DTS, CTS (=DTS+cto), composition offset, NAL types,
+	// to detect PTS non-monotonicity / gaps / param-set changes at the seam.
+	fmt.Println("=== per-sample timeline (full) — checking PTS monotonicity & nal types ===")
+	var trex *mp4ff.TrexBox
+	if parsed.Init != nil && parsed.Init.Moov != nil && parsed.Init.Moov.Mvex != nil {
+		for _, t := range parsed.Init.Moov.Mvex.Trexs {
+			if t.TrackID == 1 {
+				trex = t
+			}
+		}
+	}
+	var lastCTS int64 = -1
+	var lastDTS int64 = -1
+	sampIdx := 0
+	fragIdx = 0
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			fs, err := fr.GetFullSamples(trex)
+			if err != nil {
+				fmt.Printf("  frag%d GetFullSamples err: %v\n", fragIdx, err)
+				fragIdx++
+				continue
+			}
+			for _, s := range fs {
+				dts := int64(s.DecodeTime)
+				cts := dts + int64(s.CompositionTimeOffset)
+				nals := nalTypes(s.Data)
+				anomaly := ""
+				if lastCTS >= 0 && cts < lastCTS {
+					anomaly += fmt.Sprintf(" <== CTS BACKWARDS (prev=%d)", lastCTS)
+				}
+				if lastDTS >= 0 && dts < lastDTS {
+					anomaly += fmt.Sprintf(" <== DTS BACKWARDS (prev=%d)", lastDTS)
+				}
+				isSync := s.Flags&0x02000000 == 0 && (s.Flags>>24)&0x03 == 0x02
+				// Only print near the seam region and any anomalies, to keep output small.
+				near := dts >= 7800 && dts <= 8700
+				if near || anomaly != "" {
+					fmt.Printf("  s%04d frag%d dts=%d cts=%d cto=%d dur=%d size=%d sync=%v nal=%v%s\n",
+						sampIdx, fragIdx, dts, cts, s.CompositionTimeOffset, s.Dur, len(s.Data), isSync, nals, anomaly)
+				}
+				lastCTS = cts
+				lastDTS = dts
+				sampIdx++
+			}
+			fragIdx++
+		}
+	}
+
+	// Compare parameter sets: avcC (in moov) vs inline SPS/PPS at every IDR.
+	// A looping source that restarts may re-emit SPS/PPS that differ from the
+	// ones the player configured its decoder with from avcC — a classic cause
+	// of a freeze that "heals" when you seek past the seam.
+	fmt.Println("=== parameter set comparison (avcC vs inline IDR) ===")
+	var avccSPS, avccPPS [][]byte
+	if parsed.Init != nil && parsed.Init.Moov != nil {
+		for _, trak := range parsed.Init.Moov.Traks {
+			if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil {
+				continue
+			}
+			stsd := trak.Mdia.Minf.Stbl.Stsd
+			if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil {
+				continue
+			}
+			avccSPS = stsd.AvcX.AvcC.SPSnalus
+			avccPPS = stsd.AvcX.AvcC.PPSnalus
+		}
+	}
+	for i, s := range avccSPS {
+		fmt.Printf("  avcC SPS[%d] = %x\n", i, s)
+	}
+	for i, p := range avccPPS {
+		fmt.Printf("  avcC PPS[%d] = %x\n", i, p)
+	}
+	fragIdx = 0
+	sampIdx = 0
+	var baseSPS, basePPS []byte
+	if len(avccSPS) > 0 {
+		baseSPS = avccSPS[0]
+	}
+	if len(avccPPS) > 0 {
+		basePPS = avccPPS[0]
+	}
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			fs, err := fr.GetFullSamples(trex)
+			if err != nil {
+				fragIdx++
+				continue
+			}
+			for _, s := range fs {
+				spsList := nalsByType(s.Data, 7)
+				ppsList := nalsByType(s.Data, 8)
+				if len(spsList) > 0 || len(ppsList) > 0 {
+					dts := int64(s.DecodeTime)
+					note := ""
+					if len(spsList) > 0 {
+						if baseSPS == nil {
+							baseSPS = spsList[0]
+						} else if !bytesEqual(baseSPS, spsList[0]) {
+							note += " <== SPS CHANGED vs base/avcC"
+						}
+					}
+					if len(ppsList) > 0 {
+						if basePPS == nil {
+							basePPS = ppsList[0]
+						} else if !bytesEqual(basePPS, ppsList[0]) {
+							note += " <== PPS CHANGED vs base/avcC"
+						}
+					}
+					var spsHex, ppsHex string
+					if len(spsList) > 0 {
+						spsHex = fmt.Sprintf("%x", spsList[0])
+					}
+					if len(ppsList) > 0 {
+						ppsHex = fmt.Sprintf("%x", ppsList[0])
+					}
+					fmt.Printf("  IDR s%04d frag%d dts=%d SPS=%s PPS=%s%s\n",
+						sampIdx, fragIdx, dts, spsHex, ppsHex, note)
+				}
+				sampIdx++
+			}
+			fragIdx++
+		}
+	}
+
+	sliceHeaders(parsed, trex)
+}
+
+func sliceHeaders(parsed *mp4ff.File, trex *mp4ff.TrexBox) {
+	// Build SPS/PPS maps from avcC.
+	spsMap := map[uint32]*avc.SPS{}
+	ppsMap := map[uint32]*avc.PPS{}
+	if parsed.Init != nil && parsed.Init.Moov != nil {
+		for _, trak := range parsed.Init.Moov.Traks {
+			if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil {
+				continue
+			}
+			stsd := trak.Mdia.Minf.Stbl.Stsd
+			if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil {
+				continue
+			}
+			for _, s := range stsd.AvcX.AvcC.SPSnalus {
+				if sps, err := avc.ParseSPSNALUnit(s, true); err == nil {
+					spsMap[uint32(sps.ParameterID)] = sps
+				}
+			}
+			for _, p := range stsd.AvcX.AvcC.PPSnalus {
+				if pps, err := avc.ParsePPSNALUnit(p, spsMap); err == nil {
+					ppsMap[pps.PicParameterSetID] = pps
+				}
+			}
+		}
+	}
+
+	fmt.Println("=== slice headers near seam (frame_num / poc / idr_pic_id) ===")
+	fragIdx := 0
+	sampIdx := 0
+	for _, seg := range parsed.Segments {
+		for _, fr := range seg.Fragments {
+			fs, err := fr.GetFullSamples(trex)
+			if err != nil {
+				fragIdx++
+				continue
+			}
+			for _, s := range fs {
+				dts := int64(s.DecodeTime)
+				if dts < 6800 || dts > 9400 {
+					sampIdx++
+					continue
+				}
+				for _, nal := range splitAVCC(s.Data) {
+					t := nal[0] & 0x1f
+					if t == 1 || t == 5 { // non-IDR or IDR slice
+						sh, err := avc.ParseSliceHeader(nal, spsMap, ppsMap)
+						if err != nil {
+							fmt.Printf("  s%04d frag%d dts=%d nalType=%d sliceHeader ERR: %v\n", sampIdx, fragIdx, dts, t, err)
+							break
+						}
+						fmt.Printf("  s%04d frag%d dts=%d nalType=%d sliceType=%v frameNum=%d idrPicId=%d pocLsb=%d\n",
+							sampIdx, fragIdx, dts, t, sh.SliceType, sh.FrameNum, sh.IDRPicID, sh.PicOrderCntLsb)
+						break
+					}
+				}
+				sampIdx++
+			}
+			fragIdx++
+		}
+	}
+}
+
+// splitAVCC splits a length-prefixed (4-byte) AVCC buffer into NAL units.
+func splitAVCC(b []byte) [][]byte {
+	var out [][]byte
+	i := 0
+	for i+4 <= len(b) {
+		n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
+		i += 4
+		if n <= 0 || i+n > len(b) {
+			break
+		}
+		out = append(out, b[i:i+n])
+		i += n
+	}
+	return out
+}
+
+func bytesEqual(a, b []byte) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// nalTypes returns the list of H.264 NAL unit types present in an AVCC
+// (length-prefixed) sample buffer.
+func nalTypes(b []byte) []int {
+	var out []int
+	i := 0
+	for i+4 <= len(b) {
+		n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
+		i += 4
+		if n <= 0 || i+n > len(b) {
+			break
+		}
+		out = append(out, int(b[i]&0x1f))
+		i += n
+	}
+	return out
+}
+
+// nalsByType returns the raw NAL payloads (without length prefix) of the given
+// type from an AVCC (length-prefixed) sample buffer.
+func nalsByType(b []byte, want int) [][]byte {
+	var out [][]byte
+	i := 0
+	for i+4 <= len(b) {
+		n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3]))
+		i += 4
+		if n <= 0 || i+n > len(b) {
+			break
+		}
+		if int(b[i]&0x1f) == want {
+			nal := make([]byte, n)
+			copy(nal, b[i:i+n])
+			out = append(out, nal)
+		}
+		i += n
+	}
+	return out
+}

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
@@ -41,17 +50,6 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
 	// KerberosCloud, this means storage is disabled and proxy enabled.
 	log.Log.Info("UploadKerberosVault: Uploading to Kerberos Vault (" + config.KStorage.URI + ")")
 	log.Log.Info("UploadKerberosVault: Upload started for " + fileName)
-	fullname := "data/recordings/" + fileName
-
-	file, err := os.OpenFile(fullname, os.O_RDWR, 0755)
-	if file != nil {
-		defer file.Close()
-	}
-	if err != nil {
-		err := "UploadKerberosVault: Upload Failed, file doesn't exists anymore"
-		log.Log.Info(err)
-		return false, false, errors.New(err)
-	}
 
 	publicKey := config.KStorage.CloudKey
 	if config.HubKey != "" {
@@ -60,62 +58,30 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
 
 	// We need to check if we are in a retry timeout.
 	if kstorageRetryTimeout <= time.Now().Unix() {
+		uploaded, responded, body, err := sendToVault(*config.KStorage, publicKey, config.Key, fileName, "UploadKerberosVault", "primary")
+		if uploaded {
+			kstorageRetryCount = 0
+			log.Log.Info("UploadKerberosVault: Upload Finished, " + body)
+			return true, true, nil
+		}
 
-		req, err := http.NewRequest("POST", config.KStorage.URI+"/storage", file)
 		if err != nil {
-			errorMessage := "UploadKerberosVault: error reading request, " + config.KStorage.URI + "/storage: " + err.Error()
-			log.Log.Error(errorMessage)
-			return false, true, errors.New(errorMessage)
-		}
-		req.Header.Set("Content-Type", "video/mp4")
-		req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey)
-		req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorage.AccessKey)
-		req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorage.SecretAccessKey)
-		req.Header.Set("X-Kerberos-Storage-Provider", config.KStorage.Provider)
-		req.Header.Set("X-Kerberos-Storage-FileName", fileName)
-		req.Header.Set("X-Kerberos-Storage-Device", config.Key)
-		req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera")
-		req.Header.Set("X-Kerberos-Storage-Directory", config.KStorage.Directory)
-
-		var client *http.Client
-		if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-			tr := &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			}
-			client = &http.Client{Transport: tr}
-		} else {
-			client = &http.Client{}
-		}
-
-		resp, err := client.Do(req)
-		if resp != nil {
-			defer resp.Body.Close()
-		}
-
-		if err == nil {
-			if resp != nil {
-				body, err := io.ReadAll(resp.Body)
-				if err == nil {
-					if resp.StatusCode == 200 {
-						kstorageRetryCount = 0
-						log.Log.Info("UploadKerberosVault: Upload Finished, " + resp.Status + ", " + string(body))
-						return true, true, nil
-					} else {
-						// We increase the retry count, and set the timeout.
-						// If we have reached the retry policy, we set the timeout.
-						// This means we will not retry for the next 5 minutes.
-						if kstorageRetryCount < config.KStorage.MaxRetries {
-							kstorageRetryCount = (kstorageRetryCount + 1)
-						}
-						if kstorageRetryCount == config.KStorage.MaxRetries {
-							kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix()
-						}
-						log.Log.Info("UploadKerberosVault: Upload Failed, " + resp.Status + ", " + string(body))
-					}
-				}
-			}
-		} else {
 			log.Log.Info("UploadKerberosVault: Upload Failed, " + err.Error())
+		} else {
+			log.Log.Info("UploadKerberosVault: Upload Failed, " + body)
+		}
+
+		// We only advance the retry policy when the vault gave a definitive
+		// response (mirroring the original behaviour where transient network
+		// errors did not consume retries). When the retry count reaches the
+		// configured maximum we back off for the configured timeout.
+		if responded {
+			if kstorageRetryCount < config.KStorage.MaxRetries {
+				kstorageRetryCount = (kstorageRetryCount + 1)
+			}
+			if kstorageRetryCount == config.KStorage.MaxRetries {
+				kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix()
+			}
 		}
 	}
 
@@ -134,61 +100,116 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) (
 
 		log.Log.Info("UploadKerberosVault (Secondary): Uploading to Secondary Kerberos Vault (" + config.KStorageSecondary.URI + ")")
 
-		file, err = os.OpenFile(fullname, os.O_RDWR, 0755)
-		if file != nil {
-			defer file.Close()
-		}
-		if err != nil {
-			err := "UploadKerberosVault (Secondary): Upload Failed, file doesn't exists anymore"
-			log.Log.Info(err)
-			return false, false, errors.New(err)
+		uploaded, _, body, err := sendToVault(*config.KStorageSecondary, publicKey, config.Key, fileName, "UploadKerberosVault (Secondary)", "secondary")
+		if uploaded {
+			log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + body)
+			return true, true, nil
 		}
 
-		req, err := http.NewRequest("POST", config.KStorageSecondary.URI+"/storage", file)
 		if err != nil {
-			errorMessage := "UploadKerberosVault (Secondary): error reading request, " + config.KStorageSecondary.URI + "/storage: " + err.Error()
-			log.Log.Error(errorMessage)
-			return false, true, errors.New(errorMessage)
-		}
-		req.Header.Set("Content-Type", "video/mp4")
-		req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey)
-		req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorageSecondary.AccessKey)
-		req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorageSecondary.SecretAccessKey)
-		req.Header.Set("X-Kerberos-Storage-Provider", config.KStorageSecondary.Provider)
-		req.Header.Set("X-Kerberos-Storage-FileName", fileName)
-		req.Header.Set("X-Kerberos-Storage-Device", config.Key)
-		req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera")
-		req.Header.Set("X-Kerberos-Storage-Directory", config.KStorageSecondary.Directory)
-
-		var client *http.Client
-		if os.Getenv("AGENT_TLS_INSECURE") == "true" {
-			tr := &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			}
-			client = &http.Client{Transport: tr}
+			log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + err.Error())
 		} else {
-			client = &http.Client{}
-		}
-
-		resp, err := client.Do(req)
-		if resp != nil {
-			defer resp.Body.Close()
-		}
-
-		if err == nil {
-			if resp != nil {
-				body, err := io.ReadAll(resp.Body)
-				if err == nil {
-					if resp.StatusCode == 200 {
-						log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + resp.Status + ", " + string(body))
-						return true, true, nil
-					} else {
-						log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + resp.Status + ", " + string(body))
-					}
-				}
-			}
+			log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + body)
 		}
 	}
 
 	return false, true, nil
 }
+
+// sendToVault uploads a single recording to one Kerberos Vault. When resumable
+// uploads are enabled (the default) it attempts the tus protocol first and, if
+// the vault does not expose a tus endpoint (older deployments), transparently
+// falls back to the legacy single-shot POST.
+//
+// It returns whether the upload succeeded, whether the vault gave a definitive
+// HTTP response (so the caller can advance its retry policy), a short message
+// for logging, and a transport error if any.
+func sendToVault(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, string, error) {
+	if resumableUploadsEnabled() {
+		uploaded, responded, supported, body, err := uploadVaultResumable(vault, publicKey, deviceKey, fileName, label, slot)
+		if supported {
+			return uploaded, responded, body, err
+		}
+		log.Log.Info(label + ": resumable (tus) endpoint not available, falling back to legacy upload")
+	}
+	return uploadVaultLegacy(vault, publicKey, deviceKey, fileName, label)
+}
+
+// uploadVaultLegacy performs the original single-request upload: the whole file
+// is sent as the body of a POST to {URI}/storage. Kept for backwards
+// compatibility with vault deployments that do not support resumable uploads.
+func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, label string) (bool, bool, string, error) {
+	fullname := "data/recordings/" + fileName
+
+	file, err := os.Open(fullname)
+	if file != nil {
+		defer file.Close()
+	}
+	if err != nil {
+		msg := label + ": Upload Failed, file doesn't exists anymore"
+		log.Log.Info(msg)
+		return false, false, "", errors.New(msg)
+	}
+
+	uri := vault.URI
+	for len(uri) > 0 && uri[len(uri)-1] == '/' {
+		uri = uri[:len(uri)-1]
+	}
+
+	req, err := http.NewRequest("POST", uri+"/storage", file)
+	if err != nil {
+		errorMessage := label + ": error reading request, " + uri + "/storage: " + err.Error()
+		log.Log.Error(errorMessage)
+		return false, false, "", errors.New(errorMessage)
+	}
+	req.Header.Set("Content-Type", "video/mp4")
+	setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName)
+
+	client := newVaultHTTPClient(0)
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return false, false, "", err
+	}
+
+	body, rerr := io.ReadAll(resp.Body)
+	if rerr != nil {
+		return false, false, "", rerr
+	}
+
+	if resp.StatusCode == 200 {
+		return true, true, resp.Status + ", " + string(body), nil
+	}
+	return false, true, resp.Status + ", " + string(body), nil
+}
+
+// setVaultHeaders sets the standard Kerberos Vault headers used by the legacy
+// single-POST upload.
+func setVaultHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) {
+	h.Set("X-Kerberos-Storage-CloudKey", publicKey)
+	h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey)
+	h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey)
+	h.Set("X-Kerberos-Storage-Provider", vault.Provider)
+	h.Set("X-Kerberos-Storage-FileName", fileName)
+	h.Set("X-Kerberos-Storage-Device", deviceKey)
+	h.Set("X-Kerberos-Storage-Capture", "IPCamera")
+	h.Set("X-Kerberos-Storage-Directory", vault.Directory)
+}
+
+// newVaultHTTPClient builds an HTTP client honouring the AGENT_TLS_INSECURE
+// escape hatch. A timeout of 0 disables the client-level timeout, which is
+// required for streaming large upload bodies.
+func newVaultHTTPClient(timeout time.Duration) *http.Client {
+	client := &http.Client{}
+	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
+		client.Transport = &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+		}
+	}
+	if timeout > 0 {
+		client.Timeout = timeout
+	}
+	return client
+}

machinery/src/cloud/tus_client.go

@@ -0,0 +1,452 @@
+package cloud
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/log"
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// tusResumableVersion is the tus protocol version implemented by this client.
+const tusResumableVersion = "1.0.0"
+
+// tusUploadPath is appended to the configured Kerberos Vault URI to reach the
+// resumable upload endpoint. It mirrors how the legacy uploader appends
+// "/storage".
+const tusUploadPath = "/storage/tus/"
+
+// tusResumeState is persisted in a sidecar file next to the agent data so an
+// interrupted upload can be resumed across retries and even agent restarts.
+type tusResumeState struct {
+	UploadURL string `json:"upload_url"`
+	VaultURI  string `json:"vault_uri"`
+	Size      int64  `json:"size"`
+}
+
+// resumableUploadsEnabled reports whether the resumable (tus) upload path should
+// be attempted. It is enabled by default and can be disabled (falling back to
+// the legacy single POST) by setting AGENT_DISABLE_RESUMABLE_UPLOAD=true.
+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.
+//
+// Return values:
+//   - uploaded:  the recording was fully received and persisted by the vault.
+//   - responded: the vault returned a definitive HTTP response (used by the
+//     caller to advance its retry/secondary-failover policy).
+//   - supported: the vault exposes a tus endpoint. When false, the caller should
+//     fall back to the legacy single-POST upload (older vault deployments).
+//   - body:      a short message for logging.
+func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (uploaded bool, responded bool, supported bool, body string, err error) {
+	fullname := "data/recordings/" + fileName
+
+	file, ferr := os.Open(fullname)
+	if file != nil {
+		defer file.Close()
+	}
+	if ferr != nil {
+		msg := label + ": resumable upload failed, file doesn't exist anymore"
+		log.Log.Info(msg)
+		// The file is gone, so the legacy path cannot help either. Report it as
+		// "supported" to avoid a pointless fallback attempt.
+		return false, false, true, "", errors.New(msg)
+	}
+
+	info, serr := file.Stat()
+	if serr != nil {
+		return false, false, true, "", serr
+	}
+	size := info.Size()
+
+	baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
+	client := newVaultHTTPClient(0)
+
+	metadata := encodeTusMetadata(map[string]string{
+		"filename":  fileName,
+		"device":    deviceKey,
+		"directory": vault.Directory,
+		"provider":  vault.Provider,
+		"capture":   "IPCamera",
+		"cloudkey":  publicKey,
+	})
+
+	sidecar := tusSidecarPath(fileName, slot)
+	uploadURL := loadTusResumeState(sidecar, baseURL)
+
+	const maxAttempts = 4
+	restartedAfterComplete := false
+
+	for attempt := 0; attempt < maxAttempts; attempt++ {
+		// (1) Ensure we have an active upload URL, creating one if needed.
+		if uploadURL == "" {
+			created, status, cerr := tusCreate(client, baseURL, size, metadata, vault, publicKey, deviceKey, fileName)
+			if cerr != nil {
+				if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented {
+					// The vault does not implement tus; let the caller fall back.
+					return false, false, false, "", cerr
+				}
+				log.Log.Info(label + ": resumable create failed, " + cerr.Error())
+				tusBackoff(attempt)
+				continue
+			}
+			uploadURL = created
+			saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
+		}
+
+		// (2) Query the current server-side offset.
+		offset, status, herr := tusHead(client, uploadURL, vault, publicKey, deviceKey)
+		if herr != nil {
+			if status == http.StatusNotFound || status == http.StatusGone {
+				// The upload expired/was removed server-side; start over.
+				removeTusResumeState(sidecar)
+				uploadURL = ""
+				continue
+			}
+			log.Log.Info(label + ": resumable head failed, " + herr.Error())
+			tusBackoff(attempt)
+			continue
+		}
+
+		// (3) All bytes are present but the upload was not finalized (e.g. the
+		// completion hook failed). A completed tus upload cannot be re-finalized
+		// with another PATCH, so delete it and re-upload to force a clean finalize.
+		if offset >= size {
+			if restartedAfterComplete {
+				return false, true, true, "resumable finalize did not complete", errors.New(label + ": resumable finalize did not complete")
+			}
+			tusTerminate(client, uploadURL, vault, publicKey, deviceKey)
+			removeTusResumeState(sidecar)
+			uploadURL = ""
+			restartedAfterComplete = true
+			continue
+		}
+
+		// (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
+			}
+			continue
+		}
+
+		// 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")
+}
+
+// tusCreate performs the tus "creation" request (POST). On success it returns
+// the resolved upload URL the agent should use for subsequent HEAD/PATCH calls.
+func tusCreate(client *http.Client, baseURL string, size int64, metadata string, vault models.KStorage, publicKey, deviceKey, fileName string) (string, int, error) {
+	req, err := http.NewRequest("POST", baseURL, nil)
+	if err != nil {
+		return "", 0, err
+	}
+	req.Header.Set("Tus-Resumable", tusResumableVersion)
+	req.Header.Set("Upload-Length", strconv.FormatInt(size, 10))
+	if metadata != "" {
+		req.Header.Set("Upload-Metadata", metadata)
+	}
+	setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, fileName)
+
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return "", 0, err
+	}
+	io.Copy(io.Discard, resp.Body)
+
+	if resp.StatusCode != http.StatusCreated {
+		return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s", resp.Status)
+	}
+	location := resp.Header.Get("Location")
+	if location == "" {
+		return "", resp.StatusCode, errors.New("missing Location header in create response")
+	}
+	return resolveTusLocation(baseURL, location), resp.StatusCode, nil
+}
+
+// tusHead performs the tus "offset" request (HEAD) and returns the current
+// server-side upload offset.
+func tusHead(client *http.Client, uploadURL string, vault models.KStorage, publicKey, deviceKey string) (int64, int, error) {
+	req, err := http.NewRequest("HEAD", uploadURL, nil)
+	if err != nil {
+		return 0, 0, err
+	}
+	req.Header.Set("Tus-Resumable", tusResumableVersion)
+	setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "")
+
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return 0, 0, err
+	}
+	io.Copy(io.Discard, resp.Body)
+
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
+		return 0, resp.StatusCode, fmt.Errorf("unexpected status on HEAD: %s", resp.Status)
+	}
+	offsetStr := resp.Header.Get("Upload-Offset")
+	offset, perr := strconv.ParseInt(offsetStr, 10, 64)
+	if perr != nil {
+		return 0, resp.StatusCode, fmt.Errorf("invalid Upload-Offset header: %q", offsetStr)
+	}
+	return offset, resp.StatusCode, nil
+}
+
+// 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, 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 = 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))
+	setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "")
+
+	resp, err := client.Do(req)
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+	if err != nil {
+		return offset, 0, "", err
+	}
+	bodyBytes, _ := io.ReadAll(resp.Body)
+	respBody := string(bodyBytes)
+
+	if resp.StatusCode != http.StatusNoContent {
+		return offset, resp.StatusCode, respBody, fmt.Errorf("unexpected status on PATCH: %s, %s", resp.Status, respBody)
+	}
+	newOffsetStr := resp.Header.Get("Upload-Offset")
+	newOffset, perr := strconv.ParseInt(newOffsetStr, 10, 64)
+	if perr != 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
+}
+
+// tusTerminate best-effort deletes an upload server-side (DELETE).
+func tusTerminate(client *http.Client, uploadURL string, vault models.KStorage, publicKey, deviceKey string) {
+	req, err := http.NewRequest("DELETE", uploadURL, nil)
+	if err != nil {
+		return
+	}
+	req.Header.Set("Tus-Resumable", tusResumableVersion)
+	setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "")
+
+	resp, derr := client.Do(req)
+	if resp != nil {
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
+	}
+	_ = derr
+}
+
+// setVaultTusHeaders sets the Kerberos Vault authentication and routing headers
+// on every tus request. Credentials are sent on each request (and never stored
+// server-side in the upload metadata). When fileName is empty it is omitted, as
+// it is only useful on the creation request (routing also travels in the tus
+// Upload-Metadata).
+func setVaultTusHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) {
+	h.Set("X-Kerberos-Storage-CloudKey", publicKey)
+	h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey)
+	h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey)
+	h.Set("X-Kerberos-Storage-Provider", vault.Provider)
+	h.Set("X-Kerberos-Storage-Device", deviceKey)
+	h.Set("X-Kerberos-Storage-Directory", vault.Directory)
+	h.Set("X-Kerberos-Storage-Capture", "IPCamera")
+	if fileName != "" {
+		h.Set("X-Kerberos-Storage-FileName", fileName)
+	}
+}
+
+// encodeTusMetadata serializes a map into the tus Upload-Metadata header format:
+// a comma separated list of "key base64(value)" pairs. Keys are sorted for a
+// deterministic header value. Empty values are skipped.
+func encodeTusMetadata(pairs map[string]string) string {
+	parts := make([]string, 0, len(pairs))
+	for k, v := range pairs {
+		if v == "" {
+			continue
+		}
+		parts = append(parts, k+" "+base64.StdEncoding.EncodeToString([]byte(v)))
+	}
+	sort.Strings(parts)
+	return strings.Join(parts, ",")
+}
+
+// resolveTusLocation turns the Location header returned by the create request
+// into an absolute URL. To keep talking to the agent's configured vault host
+// (and avoid issues when the vault sits behind a proxy that rewrites the host),
+// it keeps the configured base URL and only appends the server-assigned upload
+// id taken from the Location.
+func resolveTusLocation(baseURL, location string) string {
+	if ref, err := url.Parse(location); err == nil {
+		trimmed := strings.Trim(ref.Path, "/")
+		if trimmed != "" {
+			segments := strings.Split(trimmed, "/")
+			id := segments[len(segments)-1]
+			if id != "" {
+				return strings.TrimRight(baseURL, "/") + "/" + id
+			}
+		}
+	}
+	// Fallback: resolve the reference against the base URL as-is.
+	if base, err := url.Parse(baseURL); err == nil {
+		if ref, err := url.Parse(location); err == nil {
+			return base.ResolveReference(ref).String()
+		}
+	}
+	return location
+}
+
+// tusSidecarDir is the directory where resume state files are kept. It is
+// intentionally separate from data/cloud (which is scanned for recordings to
+// upload) so the sidecar files are never mistaken for recordings.
+func tusSidecarDir() string {
+	return "data/tus"
+}
+
+func tusSidecarPath(fileName, slot string) string {
+	safe := strings.ReplaceAll(fileName, "/", "_")
+	safe = strings.ReplaceAll(safe, string(os.PathSeparator), "_")
+	return filepath.Join(tusSidecarDir(), safe+"."+slot+".json")
+}
+
+// loadTusResumeState returns a previously stored upload URL for the given
+// sidecar, but only if it was created against the same vault base URL. Any
+// mismatch or read/parse error yields an empty string (start fresh).
+func loadTusResumeState(path, baseURL string) string {
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return ""
+	}
+	var state tusResumeState
+	if err := json.Unmarshal(b, &state); err != nil {
+		return ""
+	}
+	if state.UploadURL == "" || state.VaultURI != baseURL {
+		return ""
+	}
+	return state.UploadURL
+}
+
+func saveTusResumeState(path string, state tusResumeState) {
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		return
+	}
+	b, err := json.Marshal(state)
+	if err != nil {
+		return
+	}
+	_ = os.WriteFile(path, b, 0o644)
+}
+
+func removeTusResumeState(path string) {
+	_ = os.Remove(path)
+}
+
+// tusBackoff sleeps for an exponentially increasing duration (capped) between
+// resume attempts to avoid hammering a temporarily unavailable vault.
+func tusBackoff(attempt int) {
+	delay := time.Duration(500*(1<<uint(attempt))) * time.Millisecond
+	if delay > 3*time.Second {
+		delay = 3 * time.Second
+	}
+	time.Sleep(delay)
+}

machinery/src/cloud/tus_client_test.go

@@ -0,0 +1,448 @@
+package cloud
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+// fakeUpload tracks the state of a single resumable upload on the fake server.
+type fakeUpload struct {
+	size   int64
+	offset int64
+}
+
+// fakeTus is a tiny in-memory implementation of the tus 1.0.0 server protocol,
+// sufficient to exercise the agent's resumable client.
+type fakeTus struct {
+	mu             sync.Mutex
+	uploads        map[string]*fakeUpload
+	counter        int
+	creates        int
+	lastPatchBytes int64
+	patchSizes     []int64
+
+	// unsupported makes the creation endpoint return 404, simulating an older
+	// vault without a tus endpoint.
+	unsupported bool
+	// failFinalize causes the next N completing PATCH requests to return 502
+	// after storing the bytes, simulating a failed completion hook.
+	failFinalize int
+}
+
+func newFakeTus() *fakeTus {
+	return &fakeTus{uploads: map[string]*fakeUpload{}}
+}
+
+func (s *fakeTus) seed(size, offset int64) string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.counter++
+	id := fmt.Sprintf("seed-%d", s.counter)
+	s.uploads[id] = &fakeUpload{size: size, offset: offset}
+	return id
+}
+
+func (s *fakeTus) totalBytes() int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var total int64
+	for _, u := range s.uploads {
+		total += u.offset
+	}
+	return total
+}
+
+func (s *fakeTus) lastPatch() int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	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()
+	return s.creates
+}
+
+func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	id := strings.TrimPrefix(r.URL.Path, tusUploadPath)
+	w.Header().Set("Tus-Resumable", tusResumableVersion)
+
+	switch r.Method {
+	case http.MethodPost:
+		if s.unsupported {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		length, _ := strconv.ParseInt(r.Header.Get("Upload-Length"), 10, 64)
+		s.mu.Lock()
+		s.counter++
+		s.creates++
+		newID := fmt.Sprintf("up-%d", s.counter)
+		s.uploads[newID] = &fakeUpload{size: length}
+		s.mu.Unlock()
+		w.Header().Set("Location", tusUploadPath+newID)
+		w.WriteHeader(http.StatusCreated)
+
+	case http.MethodHead:
+		s.mu.Lock()
+		u, ok := s.uploads[id]
+		s.mu.Unlock()
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		w.Header().Set("Upload-Offset", strconv.FormatInt(u.offset, 10))
+		w.Header().Set("Upload-Length", strconv.FormatInt(u.size, 10))
+		w.WriteHeader(http.StatusOK)
+
+	case http.MethodPatch:
+		s.mu.Lock()
+		u, ok := s.uploads[id]
+		s.mu.Unlock()
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		n, _ := io.Copy(io.Discard, r.Body)
+		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 {
+			s.failFinalize--
+		}
+		offset := u.offset
+		s.mu.Unlock()
+
+		w.Header().Set("Upload-Offset", strconv.FormatInt(offset, 10))
+		if failNow {
+			// Bytes are stored but the (simulated) completion hook failed.
+			w.WriteHeader(http.StatusBadGateway)
+			return
+		}
+		w.WriteHeader(http.StatusNoContent)
+
+	case http.MethodDelete:
+		s.mu.Lock()
+		delete(s.uploads, id)
+		s.mu.Unlock()
+		w.WriteHeader(http.StatusNoContent)
+
+	default:
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}
+}
+
+// withRecording switches into a fresh temp working directory containing a
+// recording at data/recordings/<fileName>. The working directory is restored on
+// cleanup. Tests using this helper must not run in parallel.
+func withRecording(t *testing.T, fileName string, payload []byte) {
+	t.Helper()
+	dir := t.TempDir()
+	old, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("getwd: %v", err)
+	}
+	if err := os.Chdir(dir); err != nil {
+		t.Fatalf("chdir: %v", err)
+	}
+	t.Cleanup(func() { _ = os.Chdir(old) })
+
+	if err := os.MkdirAll("data/recordings", 0o755); err != nil {
+		t.Fatalf("mkdir recordings: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join("data/recordings", fileName), payload, 0o644); err != nil {
+		t.Fatalf("write recording: %v", err)
+	}
+}
+
+func testVault(uri string) models.KStorage {
+	return models.KStorage{
+		URI:             uri,
+		AccessKey:       "ak",
+		SecretAccessKey: "sk",
+		Provider:        "gcp",
+		Directory:       "dir",
+	}
+}
+
+func TestUploadVaultResumable_HappyPath(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("x"), 4096)
+	withRecording(t, fileName, payload)
+
+	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !uploaded || !responded || !supported {
+		t.Fatalf("uploaded/responded/supported = %v/%v/%v, want all true", uploaded, responded, supported)
+	}
+	if got := srv.totalBytes(); got != int64(len(payload)) {
+		t.Fatalf("server received %d bytes, want %d", got, len(payload))
+	}
+	if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) {
+		t.Fatalf("expected sidecar to be removed after success, stat err = %v", err)
+	}
+}
+
+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
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "f.mp4"
+	withRecording(t, fileName, []byte("hello"))
+
+	uploaded, _, supported, _, _ := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
+	if uploaded {
+		t.Fatal("expected uploaded=false against a vault without a tus endpoint")
+	}
+	if supported {
+		t.Fatal("expected supported=false so the caller falls back to the legacy upload")
+	}
+}
+
+func TestUploadVaultResumable_FinalizeRetry(t *testing.T) {
+	srv := newFakeTus()
+	srv.failFinalize = 1
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	payload := bytes.Repeat([]byte("y"), 2048)
+	withRecording(t, fileName, payload)
+
+	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 after a failed finalize + restart, got uploaded=%v supported=%v", uploaded, supported)
+	}
+	if got := srv.createCount(); got < 2 {
+		t.Fatalf("expected at least 2 create requests (restart after failed finalize), got %d", got)
+	}
+}
+
+func TestUploadVaultResumable_ResumeFromSidecar(t *testing.T) {
+	srv := newFakeTus()
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
+	total := 8192
+	half := 4096
+	payload := bytes.Repeat([]byte("z"), total)
+	withRecording(t, fileName, payload)
+
+	// Simulate a previous run that uploaded half the file before being interrupted.
+	id := srv.seed(int64(total), int64(half))
+	baseURL := strings.TrimRight(ts.URL, "/") + tusUploadPath
+	saveTusResumeState(tusSidecarPath(fileName, "primary"), tusResumeState{
+		UploadURL: strings.TrimRight(baseURL, "/") + "/" + id,
+		VaultURI:  baseURL,
+		Size:      int64(total),
+	})
+
+	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 resume success, got uploaded=%v supported=%v", uploaded, supported)
+	}
+	if got := srv.lastPatch(); got != int64(total-half) {
+		t.Fatalf("resume should only send the remaining %d bytes, sent %d", total-half, got)
+	}
+	if srv.createCount() != 0 {
+		t.Fatalf("resume should not create a new upload, got %d creates", srv.createCount())
+	}
+}
+
+func TestEncodeTusMetadata(t *testing.T) {
+	got := encodeTusMetadata(map[string]string{
+		"b":     "2",
+		"a":     "1",
+		"empty": "",
+	})
+	// keys sorted, empty values skipped, values base64-encoded.
+	want := "a MQ==,b Mg=="
+	if got != want {
+		t.Fatalf("encodeTusMetadata = %q, want %q", got, want)
+	}
+}
+
+func TestResolveTusLocation(t *testing.T) {
+	cases := []struct {
+		name     string
+		base     string
+		location string
+		want     string
+	}{
+		{
+			name:     "absolute path location",
+			base:     "http://host/storage/tus/",
+			location: "/storage/tus/abc",
+			want:     "http://host/storage/tus/abc",
+		},
+		{
+			name:     "absolute url keeps configured host",
+			base:     "http://host/storage/tus/",
+			location: "http://internal:8080/storage/tus/xyz",
+			want:     "http://host/storage/tus/xyz",
+		},
+		{
+			name:     "relative id",
+			base:     "http://host/api/storage/tus/",
+			location: "abc",
+			want:     "http://host/api/storage/tus/abc",
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := resolveTusLocation(tc.base, tc.location); got != tc.want {
+				t.Fatalf("resolveTusLocation(%q, %q) = %q, want %q", tc.base, tc.location, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestTusResumeStateRoundTrip(t *testing.T) {
+	dir := t.TempDir()
+	old, _ := os.Getwd()
+	if err := os.Chdir(dir); err != nil {
+		t.Fatalf("chdir: %v", err)
+	}
+	defer os.Chdir(old)
+
+	path := tusSidecarPath("file.mp4", "primary")
+	state := tusResumeState{UploadURL: "http://host/storage/tus/abc", VaultURI: "http://host/storage/tus/", Size: 123}
+	saveTusResumeState(path, state)
+
+	if got := loadTusResumeState(path, state.VaultURI); got != state.UploadURL {
+		t.Fatalf("loadTusResumeState = %q, want %q", got, state.UploadURL)
+	}
+	// A mismatched vault URI must not be reused.
+	if got := loadTusResumeState(path, "http://other/storage/tus/"); got != "" {
+		t.Fatalf("loadTusResumeState with mismatched vault = %q, want empty", got)
+	}
+}

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