diff --git a/api/docs/docs.go b/api/docs/docs.go index ad6ce2c..7bc527a 100644 --- a/api/docs/docs.go +++ b/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": { diff --git a/api/docs/swagger.json b/api/docs/swagger.json index 6cc7791..fc85570 100644 --- a/api/docs/swagger.json +++ b/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": { diff --git a/api/docs/swagger.yaml b/api/docs/swagger.yaml index c8e513c..11e1366 100644 --- a/api/docs/swagger.yaml +++ b/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: diff --git a/internal/ingestor/probe_test.go b/internal/ingestor/probe_test.go new file mode 100644 index 0000000..50fd961 --- /dev/null +++ b/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)) + } +} diff --git a/internal/ingestor/service.go b/internal/ingestor/service.go index c80937c..cfc5909 100644 --- a/internal/ingestor/service.go +++ b/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. diff --git a/internal/manager/service.go b/internal/manager/service.go index 21bba79..ad84c75 100644 --- a/internal/manager/service.go +++ b/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.