From 7009884f70964689fb9e40b1fe16d8694c938ac8 Mon Sep 17 00:00:00 2001 From: ntthuan060102github Date: Thu, 18 Jun 2026 11:27:48 +0700 Subject: [PATCH] fix(failover): require a sustained stream before failing back to a recovered input MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The failback probe declared a degraded input "recovered" on a single payload packet from one read. An intermittent source that merely sputters (an occasional packet while effectively down) passed that check, so the manager failed back to it, then starved to the packet-timeout and degraded again — flapping between the source and its backup roughly every timeout window, and the repeated input switches churned ingest sessions into visible playback flicker on recovery. Probe now requires payload to keep arriving across a minimum span (1.5s), with a minimum packet count and byte total, before reporting recovery. A sputtering feed never reaches the span and stays degraded; a genuinely recovered source streams continuously and is adopted once. The manager's probe deadline is raised to 5s to fit the sustained window. --- api/docs/docs.go | 16 +++---- api/docs/swagger.json | 16 +++---- api/docs/swagger.yaml | 12 ++--- internal/ingestor/probe_test.go | 66 ++++++++++++++++++++++++++ internal/ingestor/service.go | 83 +++++++++++++++++++++++++++++---- internal/manager/service.go | 2 +- 6 files changed, 162 insertions(+), 33 deletions(-) create mode 100644 internal/ingestor/probe_test.go 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.