fix(failover): require a sustained stream before failing back to a recovered input

api/docs/docs.go

@@ -2650,8 +2650,6 @@ const docTemplate = `{
         "domain.EventType": {
             "type": "string",
             "enum": [
-                "session.opened",
-                "session.closed",
                 "stream.created",
                 "stream.updated",
                 "stream.started",
@@ -2688,7 +2686,9 @@ const docTemplate = `{
                 "policy.updated",
                 "policy.deleted",
                 "stream.runtime_created",
-                "stream.runtime_expired"
+                "stream.runtime_expired",
+                "session.opened",
+                "session.closed"
             ],
             "x-enum-comments": {
                 "EventDVRSegmentPruned": "retention loop deleted an aged-out segment",
@@ -2705,8 +2705,6 @@ const docTemplate = `{
                 "EventStreamUpdated": "PUT /streams/{code} on existing record"
             },
             "x-enum-descriptions": [
-                "",
-                "",
                 "",
                 "PUT /streams/{code} on existing record",
                 "",
@@ -2743,11 +2741,11 @@ const docTemplate = `{
                 "",
                 "",
                 "",
+                "",
+                "",
                 ""
             ],
             "x-enum-varnames": [
-                "EventSessionOpened",
-                "EventSessionClosed",
                 "EventStreamCreated",
                 "EventStreamUpdated",
                 "EventStreamStarted",
@@ -2784,7 +2782,9 @@ const docTemplate = `{
                 "EventPolicyUpdated",
                 "EventPolicyDeleted",
                 "EventStreamRuntimeCreated",
-                "EventStreamRuntimeExpired"
+                "EventStreamRuntimeExpired",
+                "EventSessionOpened",
+                "EventSessionClosed"
             ]
         },
         "domain.GlobalConfig": {

api/docs/swagger.json

@@ -2643,8 +2643,6 @@
         "domain.EventType": {
             "type": "string",
             "enum": [
-                "session.opened",
-                "session.closed",
                 "stream.created",
                 "stream.updated",
                 "stream.started",
@@ -2681,7 +2679,9 @@
                 "policy.updated",
                 "policy.deleted",
                 "stream.runtime_created",
-                "stream.runtime_expired"
+                "stream.runtime_expired",
+                "session.opened",
+                "session.closed"
             ],
             "x-enum-comments": {
                 "EventDVRSegmentPruned": "retention loop deleted an aged-out segment",
@@ -2698,8 +2698,6 @@
                 "EventStreamUpdated": "PUT /streams/{code} on existing record"
             },
             "x-enum-descriptions": [
-                "",
-                "",
                 "",
                 "PUT /streams/{code} on existing record",
                 "",
@@ -2736,11 +2734,11 @@
                 "",
                 "",
                 "",
+                "",
+                "",
                 ""
             ],
             "x-enum-varnames": [
-                "EventSessionOpened",
-                "EventSessionClosed",
                 "EventStreamCreated",
                 "EventStreamUpdated",
                 "EventStreamStarted",
@@ -2777,7 +2775,9 @@
                 "EventPolicyUpdated",
                 "EventPolicyDeleted",
                 "EventStreamRuntimeCreated",
-                "EventStreamRuntimeExpired"
+                "EventStreamRuntimeExpired",
+                "EventSessionOpened",
+                "EventSessionClosed"
             ]
         },
         "domain.GlobalConfig": {

api/docs/swagger.yaml

@@ -687,8 +687,6 @@ definitions:
     type: object
   domain.EventType:
     enum:
-    - session.opened
-    - session.closed
     - stream.created
     - stream.updated
     - stream.started
@@ -726,6 +724,8 @@ definitions:
     - policy.deleted
     - stream.runtime_created
     - stream.runtime_expired
+    - session.opened
+    - session.closed
     type: string
     x-enum-comments:
       EventDVRSegmentPruned: retention loop deleted an aged-out segment
@@ -743,8 +743,6 @@ definitions:
       EventStreamUpdated: PUT /streams/{code} on existing record
     x-enum-descriptions:
     - ""
-    - ""
-    - ""
     - PUT /streams/{code} on existing record
     - ""
     - ""
@@ -781,9 +779,9 @@ definitions:
     - ""
     - ""
     - ""
+    - ""
+    - ""
     x-enum-varnames:
-    - EventSessionOpened
-    - EventSessionClosed
     - EventStreamCreated
     - EventStreamUpdated
     - EventStreamStarted
@@ -821,6 +819,8 @@ definitions:
     - EventPolicyDeleted
     - EventStreamRuntimeCreated
     - EventStreamRuntimeExpired
+    - EventSessionOpened
+    - EventSessionClosed
   domain.GlobalConfig:
     properties:
       auth:

internal/ingestor/probe_test.go

@@ -0,0 +1,66 @@
+package ingestor
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ntt0601zcoder/open-streamer/internal/domain"
+)
+
+func TestSustainedProbeOK(t *testing.T) {
+	cases := []struct {
+		name           string
+		bytes, packets int
+		span           time.Duration
+		want           bool
+	}{
+		{"silent source", 0, 0, 0, false},
+		{"single blip", 3000, 2, 0, false},
+		{"two sparse blips far apart", 4000, 2, 2 * time.Second, false}, // span ok, but too few packets/bytes
+		{"burst then quiet (no span)", 2_000_000, 500, 0, false},        // lots of data, all at one instant
+		{"sustained healthy stream", 2_000_000, 500, 1500 * time.Millisecond, true},
+		{"boundary exact", minProbeBytes, minProbePackets, minProbeSpan, true},
+		{"bytes one below", minProbeBytes - 1, minProbePackets, minProbeSpan, false},
+		{"packets one below", minProbeBytes, minProbePackets - 1, minProbeSpan, false},
+		{"span one below", minProbeBytes, minProbePackets, minProbeSpan - time.Millisecond, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := sustainedProbeOK(tc.bytes, tc.packets, tc.span); got != tc.want {
+				t.Fatalf("sustainedProbeOK(%d,%d,%s) = %v, want %v", tc.bytes, tc.packets, tc.span, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestProbeStats_SputteringNotSustained(t *testing.T) {
+	// A sputtering feed: one small burst, then long silence, then one more —
+	// never enough packets/bytes within a continuous span. Must stay un-sustained.
+	var s probeStats
+	t0 := time.Unix(1_700_000_000, 0)
+	s.add([]domain.AVPacket{{Data: make([]byte, 1316)}, {Data: make([]byte, 1316)}}, t0)
+	if s.sustained() {
+		t.Fatal("a 2-packet blip must not count as sustained")
+	}
+	s.add([]domain.AVPacket{{Data: make([]byte, 1316)}}, t0.Add(2*time.Second))
+	if s.sustained() {
+		t.Fatal("two sparse blips 2s apart must not count as sustained")
+	}
+}
+
+func TestProbeStats_SustainedStream(t *testing.T) {
+	// Continuous payload across > minProbeSpan with plenty of packets/bytes.
+	var s probeStats
+	t0 := time.Unix(1_700_000_000, 0)
+	for i := 0; i < 20; i++ {
+		batch := make([]domain.AVPacket, 8)
+		for j := range batch {
+			batch[j] = domain.AVPacket{Data: make([]byte, 1316)}
+		}
+		s.add(batch, t0.Add(time.Duration(i)*100*time.Millisecond))
+	}
+	if !s.sustained() {
+		t.Fatalf("continuous stream should be sustained (bytes=%d packets=%d span=%s)",
+			s.bytes, s.packets, s.lastAt.Sub(s.firstAt))
+	}
+}

internal/ingestor/service.go

@@ -17,6 +17,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"golang.org/x/sync/errgroup"
 
@@ -361,6 +362,67 @@ func (s *Service) Start(ctx context.Context, streamID domain.StreamCode, input d
 
 // Probe performs a short pull-read health probe for one input.
 // It is used by the manager to verify degraded inputs before failback.
+// Probe sustained-recovery thresholds. A degraded input is only declared
+// recovered when it delivers a *sustained* stream — not a single packet. A
+// source that merely sputters (an intermittent multicast feed emitting an
+// occasional blip while still effectively down) would pass a one-packet check,
+// trigger a failback, then starve to the packet-timeout and degrade again —
+// flapping between the source and its backup every ~timeout seconds. Requiring
+// payload to keep arriving across minProbeSpan kills that blip: a real stream
+// streams continuously, a sputtering one does not.
+const (
+	minProbeSpan    = 1500 * time.Millisecond // payload must keep arriving for at least this long
+	minProbePackets = 10                      // and total at least this many payload packets
+	minProbeBytes   = 16 * 1024               // and this many payload bytes
+)
+
+// sustainedProbeOK reports whether the observed payload constitutes a sustained
+// live stream rather than a transient blip. Pure for testability.
+func sustainedProbeOK(payloadBytes, payloadPackets int, span time.Duration) bool {
+	return payloadPackets >= minProbePackets &&
+		payloadBytes >= minProbeBytes &&
+		span >= minProbeSpan
+}
+
+// probeStats accumulates payload observed during a Probe and decides whether the
+// source looks sustained. Payload "span" is measured by arrival time, so a
+// sputtering feed (a burst then quiet) never reaches minProbeSpan.
+type probeStats struct {
+	bytes, packets  int
+	firstAt, lastAt time.Time
+}
+
+func (a *probeStats) add(batch []domain.AVPacket, now time.Time) {
+	for _, p := range batch {
+		if len(p.Data) == 0 {
+			continue
+		}
+		a.bytes += len(p.Data)
+		a.packets++
+		if a.firstAt.IsZero() {
+			a.firstAt = now
+		}
+		a.lastAt = now
+	}
+}
+
+func (a *probeStats) sustained() bool {
+	return sustainedProbeOK(a.bytes, a.packets, a.lastAt.Sub(a.firstAt))
+}
+
+func (a *probeStats) notSustainedErr(cause error) error {
+	if a.packets == 0 {
+		return fmt.Errorf("ingestor: probe got no payload: %w", cause)
+	}
+	return fmt.Errorf("ingestor: probe source not sustained (%d pkts, %d bytes over %s)",
+		a.packets, a.bytes, a.lastAt.Sub(a.firstAt))
+}
+
+// Probe reports whether a (degraded) input source is deliverable right now.
+// It returns nil only when the source produces a *sustained* stream within ctx;
+// a source that is silent or merely sputtering yields an error so the manager
+// does not fail back to it prematurely. The caller's ctx deadline must exceed
+// minProbeSpan.
 func (s *Service) Probe(ctx context.Context, input domain.Input) error {
 	if protocol.IsPushListen(input.URL) {
 		return fmt.Errorf("ingestor: probe unsupported for push-listen input %q", input.URL)
@@ -374,19 +436,20 @@ func (s *Service) Probe(ctx context.Context, input domain.Input) error {
 	if err := reader.Open(ctx); err != nil {
 		return err
 	}
-	batch, err := reader.ReadPackets(ctx)
-	if err != nil {
-		return err
-	}
-	if len(batch) == 0 {
-		return fmt.Errorf("ingestor: probe got empty batch")
-	}
-	for _, p := range batch {
-		if len(p.Data) > 0 {
+
+	// Read until the source proves it is sustained (success) or ctx expires
+	// (the caller's probe deadline must exceed minProbeSpan).
+	var stats probeStats
+	for {
+		batch, rerr := reader.ReadPackets(ctx)
+		stats.add(batch, time.Now())
+		if stats.sustained() {
 			return nil
 		}
+		if rerr != nil {
+			return stats.notSustainedErr(rerr)
+		}
 	}
-	return fmt.Errorf("ingestor: probe got no payload")
 }
 
 // Stop cancels the pull worker or unregisters the push key for streamID.

internal/manager/service.go

@@ -47,7 +47,7 @@ const (
 	monitorInterval        = 2 * time.Second
 	failbackProbeCooldown  = 8 * time.Second  // min time after degradation before first probe
 	failbackSwitchCooldown = 12 * time.Second // min time between input switches
-	probeTimeout           = 3 * time.Second
+	probeTimeout           = 5 * time.Second  // must exceed ingestor minProbeSpan (sustained-stream check) + connect time
 )
 
 // InputHealth tracks the runtime health of one input source.