Merge pull request #308 from kerberos-io/feat/recording-fps-metadata

feat: propagate recording FPS with uploads

machinery/src/capture/cleanup_test.go

@@ -28,7 +28,8 @@ func writeRecording(t *testing.T, recordingsDir, name string, ageMinutes int) {
 // marking it as still queued for upload.
 func markPending(t *testing.T, cloudDir, name string) {
 	t.Helper()
-	if err := os.WriteFile(filepath.Join(cloudDir, name), nil, 0o644); err != nil {
+	markerName := models.RecordingUploadMetadataFileName(name)
+	if err := os.WriteFile(filepath.Join(cloudDir, markerName), nil, 0o644); err != nil {
 		t.Fatalf("write marker %s: %v", name, err)
 	}
 }
@@ -71,6 +72,24 @@ func TestPickRecordingToCleanup_PrefersUploaded(t *testing.T) {
 	}
 }
 
+func TestPickRecordingToCleanup_RecognizesLegacyMarkerName(t *testing.T) {
+	recordingsDir, cloudDir := newCleanupDirs(t)
+
+	writeRecording(t, recordingsDir, "legacy_pending.mp4", 30)
+	if err := os.WriteFile(filepath.Join(cloudDir, "legacy_pending.mp4"), nil, 0o644); err != nil {
+		t.Fatalf("write legacy marker: %v", err)
+	}
+	writeRecording(t, recordingsDir, "uploaded.mp4", 10)
+
+	name, pending, err := pickRecordingToCleanup(recordingsDir, cloudDir)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if pending || name != "uploaded.mp4" {
+		t.Fatalf("cleanup picked name=%q pending=%v, want uploaded.mp4 pending=false", name, pending)
+	}
+}
+
 // Among several already-uploaded recordings, the oldest uploaded one is chosen.
 func TestPickRecordingToCleanup_OldestUploadedFirst(t *testing.T) {
 	recordingsDir, cloudDir := newCleanupDirs(t)

machinery/src/capture/main.go

@@ -4,8 +4,11 @@ package capture
 import (
 	"context"
 	"encoding/base64"
+	"encoding/json"
 	"image"
+	"math"
 	"os"
+	"path/filepath"
 	"strconv"
 	"time"
 
@@ -50,6 +53,52 @@ func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration
 	}
 }
 
+func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *video.MP4) models.RecordingUploadMetadata {
+	metadata := models.RecordingUploadMetadata{
+		FileName:  filepath.Base(name),
+		DeviceKey: deviceKey,
+		Timestamp: timestamp,
+		Duration:  mp4Video.VideoTotalDuration,
+	}
+	value := mp4Video.AverageFPS()
+	if value > 0 && value <= 240 && !math.IsInf(value, 0) && !math.IsNaN(value) {
+		metadata.FPS = int(math.Floor(value))
+	}
+	return metadata
+}
+
+// queueRecordingForUpload creates the marker consumed by the upload worker and
+// stores metadata captured from the finalized recording.
+func queueRecordingForUpload(configDirectory string, metadata models.RecordingUploadMetadata) {
+	payload, err := json.Marshal(metadata)
+	if err != nil {
+		log.Log.Error("capture.main.queueRecordingForUpload(): " + err.Error())
+		return
+	}
+
+	// Publish the marker with a same-filesystem rename. Writing directly to the
+	// watched directory would briefly expose an empty file to the upload poller.
+	marker, err := os.CreateTemp(filepath.Join(configDirectory, "data"), ".upload-marker-*")
+	if err == nil {
+		_, err = marker.Write(payload)
+	}
+	if err == nil {
+		err = marker.Chmod(0644)
+	}
+	if marker != nil {
+		if closeErr := marker.Close(); err == nil {
+			err = closeErr
+		}
+		defer os.Remove(marker.Name())
+	}
+	if err == nil {
+		err = os.Rename(marker.Name(), filepath.Join(configDirectory, "data", "cloud", models.RecordingUploadMetadataFileName(metadata.FileName)))
+	}
+	if err != nil {
+		log.Log.Error("capture.main.queueRecordingForUpload(): " + err.Error())
+	}
+}
+
 const (
 	// manualRecordingHeartbeatTimeout is how long the agent keeps a manual
 	// (live-view / remote) recording alive after the LAST viewer heartbeat. The
@@ -152,8 +201,10 @@ func CleanupRecordingDirectory(configDirectory string, configuration *models.Con
 		// now-dangling upload marker so the upload loop doesn't keep trying to
 		// upload a file that no longer exists.
 		log.Log.Warning("HandleRecordStream: removed oldest recording as part of cleanup, but it was STILL PENDING UPLOAD (disk full of un-uploaded recordings) - " + recordingsDirectory + "/" + name)
-		if err := os.Remove(cloudDirectory + "/" + name); err != nil && !os.IsNotExist(err) {
-			log.Log.Info("HandleRecordStream: could not remove dangling upload marker " + name + ", " + err.Error())
+		for _, markerName := range uploadMarkerNames(name) {
+			if err := os.Remove(filepath.Join(cloudDirectory, markerName)); err != nil && !os.IsNotExist(err) {
+				log.Log.Info("HandleRecordStream: could not remove dangling upload marker " + markerName + ", " + err.Error())
+			}
 		}
 	} else {
 		log.Log.Info("HandleRecordStream: removed oldest file as part of cleanup - " + recordingsDirectory + "/" + name)
@@ -246,9 +297,9 @@ func pickRecordingToCleanup(recordingsDirectory, cloudDirectory string) (string,
 			oldestAnyTime = modTime
 		}
 
-		// A recording is still pending upload if a marker with the same name
-		// exists in the cloud directory. Skip those when picking a safe candidate.
-		if _, statErr := os.Stat(cloudDirectory + "/" + entry.Name()); statErr == nil {
+		// A recording is still pending upload if either its current .metadata
+		// marker or a marker created by an older agent exists.
+		if recordingPendingUpload(cloudDirectory, entry.Name()) {
 			continue
 		}
 
@@ -267,6 +318,19 @@ func pickRecordingToCleanup(recordingsDirectory, cloudDirectory string) (string,
 	return "", false, os.ErrNotExist
 }
 
+func uploadMarkerNames(recordingName string) []string {
+	return []string{models.RecordingUploadMetadataFileName(recordingName), filepath.Base(recordingName)}
+}
+
+func recordingPendingUpload(cloudDirectory, recordingName string) bool {
+	for _, markerName := range uploadMarkerNames(recordingName) {
+		if _, err := os.Stat(filepath.Join(cloudDirectory, markerName)); err == nil {
+			return true
+		}
+	}
+	return false
+}
+
 func HandleRecordStream(queue *packets.Queue, configDirectory string, configuration *models.Configuration, communication *models.Communication, rtspClient RTSPClient, mqttClient mqtt.Client) {
 
 	config := configuration.Config
@@ -438,9 +502,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					// Create a symbol link.
-					fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-					fc.Close()
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
 
 					recordingStatus = "idle"
 
@@ -597,9 +659,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					// Create a symbol link.
-					fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-					fc.Close()
+					queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video))
 
 					recordingStatus = "idle"
 
@@ -869,9 +929,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					}
 				}
 
-				// Create a symbol linc.
-				fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-				fc.Close()
+				queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, displayTime, mp4Video))
 
 				// Clean up the recording directory if necessary.
 				CleanupRecordingDirectory(configDirectory, configuration)

machinery/src/capture/main_test.go

@@ -0,0 +1,60 @@
+package capture
+
+import (
+	"encoding/json"
+	"math"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/agent/machinery/src/video"
+)
+
+func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) {
+	configDirectory := t.TempDir()
+	if err := os.MkdirAll(filepath.Join(configDirectory, "data", "cloud"), 0o755); err != nil {
+		t.Fatalf("mkdir cloud queue: %v", err)
+	}
+
+	mp4Video := &video.MP4{VideoTotalDuration: 20452, SampleCount: 613}
+	metadata := recordingUploadMetadata("recording.mp4", "device-key", 1785934709414, mp4Video)
+	queueRecordingForUpload(configDirectory, metadata)
+
+	got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata"))
+	if err != nil {
+		t.Fatalf("read upload marker: %v", err)
+	}
+	var stored models.RecordingUploadMetadata
+	if err := json.Unmarshal(got, &stored); err != nil {
+		t.Fatalf("decode upload marker: %v", err)
+	}
+	if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || stored.FPS != 29 {
+		t.Fatalf("upload marker = %+v", stored)
+	}
+}
+
+func TestQueueRecordingForUploadKeepsUnknownFPSCompatible(t *testing.T) {
+	for _, fps := range []float64{0, 0.99, -1, math.NaN(), math.Inf(1), 241} {
+		t.Run("invalid FPS", func(t *testing.T) {
+			configDirectory := t.TempDir()
+			if err := os.MkdirAll(filepath.Join(configDirectory, "data", "cloud"), 0o755); err != nil {
+				t.Fatalf("mkdir cloud queue: %v", err)
+			}
+
+			metadata := models.RecordingUploadMetadata{FileName: "recording.mp4"}
+			if fps >= 1 && fps <= 240 && !math.IsNaN(fps) && !math.IsInf(fps, 0) {
+				metadata.FPS = int(math.Floor(fps))
+			}
+			queueRecordingForUpload(configDirectory, metadata)
+
+			got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata"))
+			if err != nil {
+				t.Fatalf("read upload marker: %v", err)
+			}
+			if string(got) != `{"filename":"recording.mp4","device_key":"","timestamp":0,"duration":0}` {
+				t.Fatalf("upload marker = %q, want metadata without FPS", got)
+			}
+		})
+	}
+}

machinery/src/cloud/cloud.go

@@ -78,7 +78,8 @@ func HandleUpload(configDirectory string, configuration *models.Configuration, c
 					default:
 					}
 
-					fileName := f.Name()
+					markerFileName := f.Name()
+					fileName := models.RecordingFileNameFromUploadMarker(markerFileName)
 					uploaded := false
 					configured := false
 					err = nil
@@ -113,7 +114,7 @@ func HandleUpload(configDirectory string, configuration *models.Configuration, c
 					// Check if the file is uploaded, if so, remove it.
 					if uploaded {
 						delay = 500 * time.Millisecond // reset
-						err := os.Remove(watchDirectory + fileName)
+						err := os.Remove(watchDirectory + markerFileName)
 						if err != nil {
 							log.Log.Error("HandleUpload: " + err.Error())
 						}
@@ -127,7 +128,7 @@ func HandleUpload(configDirectory string, configuration *models.Configuration, c
 							}
 						}
 					} else if !configured {
-						err := os.Remove(watchDirectory + fileName)
+						err := os.Remove(watchDirectory + markerFileName)
 						if err != nil {
 							log.Log.Error("HandleUpload: " + err.Error())
 						}

machinery/src/cloud/kerberos_hub.go

@@ -85,6 +85,7 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 	req.Header.Set("X-Kerberos-Hub-PublicKey", config.HubKey)
 	req.Header.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey)
 	req.Header.Set("X-Kerberos-Hub-Region", config.S3.Region)
+	setQueuedRecordingMetadataHeaders(req.Header, fileName)
 
 	var client *http.Client
 	if os.Getenv("AGENT_TLS_INSECURE") == "true" {
@@ -128,6 +129,7 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo
 	req.Header.Set("X-Kerberos-Hub-PublicKey", config.HubKey)
 	req.Header.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey)
 	req.Header.Set("X-Kerberos-Hub-Region", config.S3.Region)
+	setQueuedRecordingMetadataHeaders(req.Header, fileName)
 	resp, err = client.Do(req)
 	if resp != nil {
 		defer resp.Body.Close()

machinery/src/cloud/kerberos_vault.go

@@ -165,6 +165,7 @@ func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, la
 	}
 	req.Header.Set("Content-Type", "video/mp4")
 	setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName)
+	setQueuedRecordingMetadataHeaders(req.Header, fileName)
 
 	client := newVaultHTTPClient(0)
 	resp, err := client.Do(req)

machinery/src/cloud/recording_metadata.go

@@ -0,0 +1,88 @@
+package cloud
+
+import (
+	"encoding/json"
+	"math"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+const recordingFPSHeader = "X-Kerberos-Storage-Fps"
+const recordingDurationHeader = "X-Kerberos-Storage-Duration"
+const recordingTimestampHeader = "X-Kerberos-Storage-Timestamp"
+
+// queuedRecordingFPS reads the FPS snapshot written into the upload marker
+// when the recording was finalized. Historical empty markers intentionally
+// return no value so receivers can retain their existing MP4-derived fallback.
+func queuedRecordingFPS(fileName string) string {
+	value, ok := readRecordingUploadMetadata(fileName)
+	if !ok {
+		return ""
+	}
+
+	marker := strings.TrimSpace(string(value))
+	if strings.HasPrefix(marker, "{") {
+		metadata, ok := decodeRecordingUploadMetadata(value)
+		if !ok || metadata.FPS <= 0 || metadata.FPS > 240 {
+			return ""
+		}
+		return strconv.Itoa(metadata.FPS)
+	}
+
+	// Compatibility with markers created before upload metadata used JSON.
+	fps := marker
+	parsed, err := strconv.ParseFloat(fps, 64)
+	if err != nil || parsed <= 0 || parsed > 240 || math.IsInf(parsed, 0) || math.IsNaN(parsed) {
+		return ""
+	}
+	return fps
+}
+
+func queuedRecordingMetadata(fileName string) (models.RecordingUploadMetadata, bool) {
+	value, ok := readRecordingUploadMetadata(fileName)
+	if !ok || !strings.HasPrefix(strings.TrimSpace(string(value)), "{") {
+		return models.RecordingUploadMetadata{}, false
+	}
+	return decodeRecordingUploadMetadata(value)
+}
+
+func decodeRecordingUploadMetadata(value []byte) (models.RecordingUploadMetadata, bool) {
+	var metadata models.RecordingUploadMetadata
+	if err := json.Unmarshal(value, &metadata); err != nil {
+		return models.RecordingUploadMetadata{}, false
+	}
+	return metadata, true
+}
+
+func readRecordingUploadMetadata(fileName string) ([]byte, bool) {
+	markerNames := []string{
+		models.RecordingUploadMetadataFileName(fileName),
+		filepath.Base(fileName),
+	}
+	for _, markerName := range markerNames {
+		value, err := os.ReadFile(filepath.Join("data", "cloud", markerName))
+		if err == nil {
+			return value, true
+		}
+	}
+	return nil, false
+}
+
+func setQueuedRecordingMetadataHeaders(header http.Header, fileName string) {
+	if fps := queuedRecordingFPS(fileName); fps != "" {
+		header.Set(recordingFPSHeader, fps)
+	}
+	if metadata, ok := queuedRecordingMetadata(fileName); ok {
+		if metadata.Duration > 0 {
+			header.Set(recordingDurationHeader, strconv.FormatUint(metadata.Duration, 10))
+		}
+		if metadata.Timestamp > 0 {
+			header.Set(recordingTimestampHeader, strconv.FormatInt(metadata.Timestamp, 10))
+		}
+	}
+}

machinery/src/cloud/tus_client.go

@@ -305,14 +305,17 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 // is additionally carried in the tus Upload-Metadata.
 func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, bool, string, error) {
 	baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
-	metadata := encodeTusMetadata(map[string]string{
+	metadataValues := map[string]string{
 		"filename":  fileName,
 		"device":    deviceKey,
 		"directory": vault.Directory,
 		"provider":  vault.Provider,
 		"capture":   "IPCamera",
 		"cloudkey":  publicKey,
-	})
+		"fps":       queuedRecordingFPS(fileName),
+	}
+	addRecordingTusMetadata(metadataValues, fileName)
+	metadata := encodeTusMetadata(metadataValues)
 	setHeaders := func(h http.Header, fn string) {
 		setVaultTusHeaders(h, vault, publicKey, deviceKey, fn)
 	}
@@ -326,17 +329,33 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName,
 // intentionally omitted from the metadata here.
 func uploadHubResumable(config *models.Config, fileName, label, slot string) (bool, bool, bool, string, error) {
 	baseURL := strings.TrimRight(config.HubURI, "/") + tusUploadPath
-	metadata := encodeTusMetadata(map[string]string{
+	metadataValues := map[string]string{
 		"filename": fileName,
 		"device":   config.Key,
 		"capture":  "IPCamera",
-	})
+		"fps":      queuedRecordingFPS(fileName),
+	}
+	addRecordingTusMetadata(metadataValues, fileName)
+	metadata := encodeTusMetadata(metadataValues)
 	setHeaders := func(h http.Header, fn string) {
 		setHubTusHeaders(h, config, fn)
 	}
 	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
 }
 
+func addRecordingTusMetadata(values map[string]string, fileName string) {
+	metadata, ok := queuedRecordingMetadata(fileName)
+	if !ok {
+		return
+	}
+	if metadata.Duration > 0 {
+		values["duration"] = strconv.FormatUint(metadata.Duration, 10)
+	}
+	if metadata.Timestamp > 0 {
+		values["timestamp"] = strconv.FormatInt(metadata.Timestamp, 10)
+	}
+}
+
 // 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, setHeaders tusHeaderFunc, fileName string) (string, int, error) {

machinery/src/cloud/tus_client_test.go

@@ -243,6 +243,17 @@ func withRecording(t *testing.T, fileName string, payload []byte) {
 	}
 }
 
+func withQueuedRecordingFPS(t *testing.T, fileName, fps string) {
+	t.Helper()
+	if err := os.MkdirAll("data/cloud", 0o755); err != nil {
+		t.Fatalf("mkdir cloud queue: %v", err)
+	}
+	markerName := models.RecordingUploadMetadataFileName(fileName)
+	if err := os.WriteFile(filepath.Join("data/cloud", markerName), []byte(fps), 0o644); err != nil {
+		t.Fatalf("write cloud queue marker: %v", err)
+	}
+}
+
 func testVault(uri string) models.KStorage {
 	return models.KStorage{
 		URI:             uri,
@@ -261,6 +272,7 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
 	payload := bytes.Repeat([]byte("x"), 4096)
 	withRecording(t, fileName, payload)
+	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":29}`)
 
 	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
 	if err != nil {
@@ -275,6 +287,102 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) {
 	if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) {
 		t.Fatalf("expected sidecar to be removed after success, stat err = %v", err)
 	}
+	posts := srv.requestsForMethod(http.MethodPost)
+	metadata := decodeTusMetadata(posts[0].header.Get("Upload-Metadata"))
+	if got := metadata["fps"]; got != "29" {
+		t.Fatalf("POST metadata fps = %q, want %q", got, "29")
+	}
+	if got := metadata["duration"]; got != "20452" {
+		t.Fatalf("POST metadata duration = %q, want %q", got, "20452")
+	}
+	if got := metadata["timestamp"]; got != "1785934709414" {
+		t.Fatalf("POST metadata timestamp = %q, want %q", got, "1785934709414")
+	}
+}
+
+func TestQueuedRecordingFPSValidation(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		fps  string
+		want string
+	}{
+		{name: "json", fps: `{"fps":29}`, want: "29"},
+		{name: "json with future field", fps: `{"fps":29,"codec":"h264"}`, want: "29"},
+		{name: "json without fps", fps: `{}`},
+		{name: "json invalid fps", fps: `{"fps":241}`},
+		{name: "legacy fractional", fps: "29.97", want: "29.97"},
+		{name: "legacy trimmed", fps: " 25 \n", want: "25"},
+		{name: "empty"},
+		{name: "invalid", fps: "invalid"},
+		{name: "zero", fps: "0"},
+		{name: "negative", fps: "-1"},
+		{name: "nan", fps: "NaN"},
+		{name: "infinite", fps: "+Inf"},
+		{name: "unreasonable", fps: "241"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			fileName := "recording.mp4"
+			withRecording(t, fileName, []byte("recording"))
+			withQueuedRecordingFPS(t, fileName, test.fps)
+
+			if got := queuedRecordingFPS(fileName); got != test.want {
+				t.Fatalf("queuedRecordingFPS() = %q, want %q", got, test.want)
+			}
+
+			header := make(http.Header)
+			setQueuedRecordingMetadataHeaders(header, fileName)
+			if got := header.Get(recordingFPSHeader); got != test.want {
+				t.Fatalf("legacy FPS header = %q, want %q", got, test.want)
+			}
+		})
+	}
+}
+
+func TestQueuedRecordingFPSAllowsMissingHistoricalMarker(t *testing.T) {
+	fileName := "recording.mp4"
+	withRecording(t, fileName, []byte("recording"))
+
+	if got := queuedRecordingFPS(fileName); got != "" {
+		t.Fatalf("queuedRecordingFPS() = %q, want empty for missing marker", got)
+	}
+	header := make(http.Header)
+	setQueuedRecordingMetadataHeaders(header, fileName)
+	if got := header.Get(recordingFPSHeader); got != "" {
+		t.Fatalf("legacy FPS header = %q, want empty for missing marker", got)
+	}
+}
+
+func TestQueuedRecordingMetadataHeaders(t *testing.T) {
+	fileName := "recording.mp4"
+	withRecording(t, fileName, []byte("recording"))
+	withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":25}`)
+
+	header := make(http.Header)
+	setQueuedRecordingMetadataHeaders(header, fileName)
+	if got := header.Get(recordingFPSHeader); got != "25" {
+		t.Fatalf("FPS header = %q", got)
+	}
+	if got := header.Get(recordingDurationHeader); got != "20452" {
+		t.Fatalf("duration header = %q", got)
+	}
+	if got := header.Get(recordingTimestampHeader); got != "1785934709414" {
+		t.Fatalf("timestamp header = %q", got)
+	}
+}
+
+func TestQueuedRecordingFPSAllowsLegacyMarkerFileName(t *testing.T) {
+	fileName := "recording.mp4"
+	withRecording(t, fileName, []byte("recording"))
+	if err := os.MkdirAll("data/cloud", 0o755); err != nil {
+		t.Fatalf("mkdir cloud queue: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join("data/cloud", fileName), []byte("25"), 0o644); err != nil {
+		t.Fatalf("write legacy cloud queue marker: %v", err)
+	}
+
+	if got := queuedRecordingFPS(fileName); got != "25" {
+		t.Fatalf("queuedRecordingFPS() = %q, want legacy marker FPS", got)
+	}
 }
 
 func TestUploadVaultResumable_Chunked(t *testing.T) {
@@ -579,6 +687,7 @@ func TestUploadHubResumable_HappyPath(t *testing.T) {
 	fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4"
 	payload := bytes.Repeat([]byte("h"), 4096)
 	withRecording(t, fileName, payload)
+	withQueuedRecordingFPS(t, fileName, "29.97")
 
 	uploaded, _, supported, _, err := uploadHubResumable(testHubConfig(ts.URL), fileName, "test", "hub")
 	if err != nil {
@@ -649,6 +758,9 @@ func TestUploadHubResumable_HappyPath(t *testing.T) {
 	if meta["capture"] != "IPCamera" {
 		t.Errorf("hub metadata capture = %q, want %q", meta["capture"], "IPCamera")
 	}
+	if meta["fps"] != "29.97" {
+		t.Errorf("hub metadata fps = %q, want %q", meta["fps"], "29.97")
+	}
 }
 
 func TestUploadHubResumable_Unsupported(t *testing.T) {

machinery/src/models/recording_upload_metadata.go

@@ -0,0 +1,37 @@
+package models
+
+import (
+	"path/filepath"
+	"strings"
+)
+
+const RecordingUploadMetadataExtension = ".metadata"
+
+// RecordingUploadMetadata is persisted in the upload queue marker associated
+// with a recording. New optional fields can be added without changing the queue
+// mechanism or breaking older agents.
+type RecordingUploadMetadata struct {
+	FileName  string `json:"filename"`
+	DeviceKey string `json:"device_key"`
+	Timestamp int64  `json:"timestamp"` // Unix milliseconds.
+	Duration  uint64 `json:"duration"`  // Milliseconds.
+	FPS       int    `json:"fps,omitempty"`
+}
+
+// RecordingUploadMetadataFileName returns the queue marker name associated
+// with a recording, replacing the recording extension with .metadata.
+func RecordingUploadMetadataFileName(recordingFileName string) string {
+	name := filepath.Base(recordingFileName)
+	extension := filepath.Ext(name)
+	return strings.TrimSuffix(name, extension) + RecordingUploadMetadataExtension
+}
+
+// RecordingFileNameFromUploadMarker resolves a queue entry to its recording.
+// Markers created by older agents used the recording filename directly.
+func RecordingFileNameFromUploadMarker(markerFileName string) string {
+	name := filepath.Base(markerFileName)
+	if strings.HasSuffix(name, RecordingUploadMetadataExtension) {
+		return strings.TrimSuffix(name, RecordingUploadMetadataExtension) + ".mp4"
+	}
+	return name
+}

machinery/src/models/recording_upload_metadata_test.go

@@ -0,0 +1,15 @@
+package models
+
+import "testing"
+
+func TestRecordingUploadMetadataFileNames(t *testing.T) {
+	if got := RecordingUploadMetadataFileName("141245_x_x_.mp4"); got != "141245_x_x_.metadata" {
+		t.Fatalf("RecordingUploadMetadataFileName() = %q", got)
+	}
+	if got := RecordingFileNameFromUploadMarker("141245_x_x_.metadata"); got != "141245_x_x_.mp4" {
+		t.Fatalf("RecordingFileNameFromUploadMarker() = %q", got)
+	}
+	if got := RecordingFileNameFromUploadMarker("legacy.mp4"); got != "legacy.mp4" {
+		t.Fatalf("legacy RecordingFileNameFromUploadMarker() = %q", got)
+	}
+}

machinery/src/video/mp4.go

@@ -283,6 +283,8 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	err := mp4.MultiTrackFragment.AddFullSampleToTrack(*mp4.VideoFullSample, uint32(mp4.VideoTrack))
 	if err != nil {
 		log.Log.Error("mp4.flushPendingVideoSample(): error adding sample: " + err.Error())
+	} else {
+		mp4.SampleCount++
 	}
 	if isKF {
 		mp4.TotalKeyframesWritten++
@@ -296,6 +298,15 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
 	return true
 }
 
+// AverageFPS returns the average frame rate of the video samples actually
+// committed to this recording.
+func (mp4 *MP4) AverageFPS() float64 {
+	if mp4.SampleCount == 0 || mp4.VideoTotalDuration == 0 {
+		return 0
+	}
+	return float64(mp4.SampleCount) * 1000 / float64(mp4.VideoTotalDuration)
+}
+
 // AddSampleToTrack appends a sample to the given track.
 //
 // For video, pts is the decode timestamp (DTS, in milliseconds) and

machinery/src/video/mp4_duration_test.go

@@ -2,6 +2,7 @@ package video
 
 import (
 	"fmt"
+	"math"
 	"os"
 	"testing"
 
@@ -173,4 +174,10 @@ func TestMP4Duration(t *testing.T) {
 		t.Errorf("MISMATCH: mdhd.Duration should be 0 for fragmented MP4, got %d",
 			parsedFile.Moov.Traks[0].Mdia.Mdhd.Duration)
 	}
+	if mp4Video.SampleCount != sampleCount {
+		t.Errorf("SampleCount = %d, finalized MP4 contains %d video samples", mp4Video.SampleCount, sampleCount)
+	}
+	if fps := mp4Video.AverageFPS(); math.Abs(fps-25) > 0.001 {
+		t.Errorf("AverageFPS() = %.3f, want 25", fps)
+	}
 }