feat: propagate recording FPS with uploads

machinery/src/capture/main.go

@@ -5,8 +5,11 @@ import (
 	"context"
 	"encoding/base64"
 	"image"
+	"math"
 	"os"
+	"path/filepath"
 	"strconv"
+	"strings"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -50,6 +53,42 @@ func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration
 	}
 }
 
+// queueRecordingForUpload creates the marker consumed by the upload worker and
+// snapshots the main-stream FPS into it. Keeping the value with the recording
+// prevents a delayed upload from using the FPS of a later camera configuration.
+// Empty markers remain valid for recordings whose FPS is not yet known.
+func queueRecordingForUpload(configDirectory, name string, configuration *models.Configuration) {
+	fps := ""
+	if configuration != nil {
+		candidate := strings.TrimSpace(configuration.Config.Capture.IPCamera.FPS)
+		if parsed, err := strconv.ParseFloat(candidate, 64); err == nil && parsed > 0 && parsed <= 240 && !math.IsInf(parsed, 0) && !math.IsNaN(parsed) {
+			fps = candidate
+		}
+	}
+
+	// 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.WriteString(fps)
+	}
+	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", filepath.Base(name)))
+	}
+	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
@@ -438,9 +477,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					// Create a symbol link.
-					fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-					fc.Close()
+					queueRecordingForUpload(configDirectory, name, configuration)
 
 					recordingStatus = "idle"
 
@@ -597,9 +634,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						}
 					}
 
-					// Create a symbol link.
-					fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-					fc.Close()
+					queueRecordingForUpload(configDirectory, name, configuration)
 
 					recordingStatus = "idle"
 
@@ -869,9 +904,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					}
 				}
 
-				// Create a symbol linc.
-				fc, _ := os.Create(configDirectory + "/data/cloud/" + name)
-				fc.Close()
+				queueRecordingForUpload(configDirectory, name, configuration)
 
 				// Clean up the recording directory if necessary.
 				CleanupRecordingDirectory(configDirectory, configuration)

machinery/src/capture/main_test.go

@@ -0,0 +1,51 @@
+package capture
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+)
+
+func TestQueueRecordingForUploadSnapshotsFPS(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)
+	}
+
+	configuration := &models.Configuration{}
+	configuration.Config.Capture.IPCamera.FPS = "29.97"
+	queueRecordingForUpload(configDirectory, "recording.mp4", configuration)
+
+	got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.mp4"))
+	if err != nil {
+		t.Fatalf("read upload marker: %v", err)
+	}
+	if string(got) != "29.97" {
+		t.Fatalf("upload marker FPS = %q, want %q", got, "29.97")
+	}
+}
+
+func TestQueueRecordingForUploadKeepsUnknownFPSCompatible(t *testing.T) {
+	for _, fps := range []string{"", "invalid", "0", "NaN", "241"} {
+		t.Run(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)
+			}
+
+			configuration := &models.Configuration{}
+			configuration.Config.Capture.IPCamera.FPS = fps
+			queueRecordingForUpload(configDirectory, "recording.mp4", configuration)
+
+			got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.mp4"))
+			if err != nil {
+				t.Fatalf("read upload marker: %v", err)
+			}
+			if len(got) != 0 {
+				t.Fatalf("upload marker = %q, want empty", got)
+			}
+		})
+	}
+}

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)
+	setQueuedRecordingFPSHeader(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)
+	setQueuedRecordingFPSHeader(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)
+	setQueuedRecordingFPSHeader(req.Header, fileName)
 
 	client := newVaultHTTPClient(0)
 	resp, err := client.Do(req)

machinery/src/cloud/recording_metadata.go

@@ -0,0 +1,35 @@
+package cloud
+
+import (
+	"math"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+)
+
+const recordingFPSHeader = "X-Kerberos-Storage-Fps"
+
+// 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, err := os.ReadFile(filepath.Join("data", "cloud", filepath.Base(fileName)))
+	if err != nil {
+		return ""
+	}
+
+	fps := strings.TrimSpace(string(value))
+	parsed, err := strconv.ParseFloat(fps, 64)
+	if err != nil || parsed <= 0 || parsed > 240 || math.IsInf(parsed, 0) || math.IsNaN(parsed) {
+		return ""
+	}
+	return fps
+}
+
+func setQueuedRecordingFPSHeader(header http.Header, fileName string) {
+	if fps := queuedRecordingFPS(fileName); fps != "" {
+		header.Set(recordingFPSHeader, fps)
+	}
+}

machinery/src/cloud/tus_client.go

@@ -312,6 +312,7 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName,
 		"provider":  vault.Provider,
 		"capture":   "IPCamera",
 		"cloudkey":  publicKey,
+		"fps":       queuedRecordingFPS(fileName),
 	})
 	setHeaders := func(h http.Header, fn string) {
 		setVaultTusHeaders(h, vault, publicKey, deviceKey, fn)
@@ -330,6 +331,7 @@ func uploadHubResumable(config *models.Config, fileName, label, slot string) (bo
 		"filename": fileName,
 		"device":   config.Key,
 		"capture":  "IPCamera",
+		"fps":      queuedRecordingFPS(fileName),
 	})
 	setHeaders := func(h http.Header, fn string) {
 		setHubTusHeaders(h, config, fn)

machinery/src/cloud/tus_client_test.go

@@ -243,6 +243,16 @@ 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)
+	}
+	if err := os.WriteFile(filepath.Join("data/cloud", fileName), []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 +271,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, "29.97")
 
 	uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary")
 	if err != nil {
@@ -275,6 +286,58 @@ 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)
+	if got := decodeTusMetadata(posts[0].header.Get("Upload-Metadata"))["fps"]; got != "29.97" {
+		t.Fatalf("POST metadata fps = %q, want %q", got, "29.97")
+	}
+}
+
+func TestQueuedRecordingFPSValidation(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		fps  string
+		want string
+	}{
+		{name: "fractional", fps: "29.97", want: "29.97"},
+		{name: "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)
+			setQueuedRecordingFPSHeader(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)
+	setQueuedRecordingFPSHeader(header, fileName)
+	if got := header.Get(recordingFPSHeader); got != "" {
+		t.Fatalf("legacy FPS header = %q, want empty for missing marker", got)
+	}
 }
 
 func TestUploadVaultResumable_Chunked(t *testing.T) {
@@ -579,6 +642,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 +713,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) {