Retire orphaned undefined p/f events after cancellation grace

internal/model/eit.go

@@ -35,6 +35,8 @@ type ScheduleSnapshot struct {
 // PresentFollowingSnapshot contains both sections of one p/f subtable version.
 // Empty sections are significant: absence from both slots ends an event.
 type PresentFollowingSnapshot struct {
-	Service ServiceKey
-	Events  []Event
+	// ObservedAt is the reception time in Unix milliseconds, not event time.
+	ObservedAt int64
+	Service    ServiceKey
+	Events     []Event
 }

internal/program/EIT.md

@@ -15,9 +15,9 @@ TS の actual p/f と MH-EIT p/f について、同じ版の present/following 
 
 p/f が有効な番組の時刻・基本情報は schedule で上書きしない。同じ event_id の時刻確定は通常の更新として反映する。開始・継続時間とも未定のプレースホルダーは保存しない。
 
-present から following に移った番組は中断として保存し、両スロットから消えるまで保持する。終了が確認できた終了未定番組は、終了時刻を推定せず番組一覧から削除し、remove 通知を発行する。これは番組一覧の保存方針であり、仕様が履歴の削除を要求しているわけではない。終了時刻が既知の番組は履歴を残し、p/f の状態だけ解除する。まだ present になっていない following が消えただけでは放送終了・休止と断定しない。
+present から following に移った番組は中断として保存し、両スロットから消えるまで保持する。終了が確認できた終了未定番組は、終了時刻を推定せず番組一覧から削除し、remove 通知を発行する。これは番組一覧の保存方針であり、仕様が履歴の削除を要求しているわけではない。終了時刻が既知の番組は履歴を残し、p/f の状態だけ解除する。まだ present になっていない following が消えただけでは直ちに放送終了・休止と断定しない。ただし、同節の休止判定（当初開始予定から3時間超）を踏まえ、受信時刻が開始予定から3時間を超えた完全な p/f にも存在せず、保存時刻も未定の番組は削除する。解除済みの p/f フラグに依存せず再判定するため、過去のバージョンで残った未定番組も対象になる。現在の p/f にある番組、開始未定・未来の番組、時刻既知の履歴はこの削除の対象にしない。
 
-過去に受信した EIT の全変更履歴は保存していない。この修正で、昨晩放送された終了時刻付き更新の有無を遡及的に証明することはできない。既存の終了未定番組は対象サービスの完全な p/f を次に受信したときに整理される。
+過去に受信した EIT の全変更履歴は保存していない。この修正で、昨晩放送された終了時刻付き更新の有無を遡及的に証明することはできない。既存の終了未定番組も、対象サービスの完全な p/f を次に受信したときに同じ条件で再判定される。
 
 ## remote の番組情報
 

internal/program/present_following.go

@@ -2,6 +2,7 @@ package program
 
 import (
 	"context"
+	"time"
 
 	"github.com/21S1298001/mahiron/internal/model"
 )
@@ -27,15 +28,20 @@ func (m *Manager) reconcilePresentFollowing(ctx context.Context, programs []*Pro
 	}
 	retired := make(map[int64]bool)
 	for _, p := range stored {
-		if current[p.ID] || (!p.EITPresent && !p.EITFollowing) {
+		if current[p.ID] {
 			continue
 		}
-		// Keep known history, but do not invent an end time for an undefined event
-		// that disappeared from both slots (TR-B14 19.3, C.5.7).
-		if p.DurationOrZero() <= 0 && (p.EITPresent || p.EITSuspended) {
+		// Reconsider unlabelled orphans too: a following event can disappear
+		// before we ever observe it as present (TR-B14 19.3 cancellation grace).
+		overdue := snapshot.ObservedAt > 0 && p.StartAt != nil &&
+			*p.StartAt < snapshot.ObservedAt-int64(3*time.Hour/time.Millisecond)
+		if p.DurationOrZero() <= 0 && (p.EITPresent || p.EITSuspended || overdue) {
 			retired[p.ID] = true
 			continue
 		}
+		if !p.EITPresent && !p.EITFollowing {
+			continue
+		}
 		p.EITPresent = false
 		p.EITFollowing = false
 		p.EITSuspended = false

internal/program/present_following_test.go

@@ -2,6 +2,7 @@ package program
 
 import (
 	"context"
+	"time"
 
 	"github.com/21S1298001/mahiron/internal/model"
 	"testing"
@@ -178,3 +179,114 @@ func TestEmptyPresentFollowingRetiresOnlyItsService(t *testing.T) {
 		t.Fatalf("another service was changed: %v", err)
 	}
 }
+
+func TestPresentFollowingReconsidersOrphanedUndefinedEvents(t *testing.T) {
+	ctx := context.Background()
+	m := newTestManager(t)
+	pub := &fakeProgramEventPublisher{}
+	m.events = pub
+	key := model.ServiceKey{NetworkID: 32736, ServiceID: 1024}
+	start := int64(100000000)
+	old := model.Event{Key: key, EventID: 6473, StartAt: &start, EITFollowing: true}
+	update := func(at int64, events ...model.Event) {
+		t.Helper()
+		if err := m.UpsertPresentFollowing(ctx, model.PresentFollowingSnapshot{Service: key, ObservedAt: at, Events: events}); err != nil {
+			t.Fatal(err)
+		}
+		m.flushEvents()
+	}
+	update(start, old)
+	next := model.Event{Key: key, EventID: 4572, StartAt: testPtr(start + 1000), DurationMS: testPtr(1000), EITPresent: true}
+	// A following event disappearing is not immediate proof of cancellation.
+	update(start+2000, next)
+	p, ok, err := m.Get(ctx, model.ProgramID(key, 6473))
+	if err != nil || !ok || p.EITFollowing || p.EITPresent {
+		t.Fatalf("orphan state=%+v %v", p, err)
+	}
+	// Re-read the persisted row, just as after restarting the manager.
+	m = NewManager(m.store, pub)
+	update(start+int64((3*time.Hour)/time.Millisecond), next)
+	if _, ok, err := m.Get(ctx, model.ProgramID(key, 6473)); err != nil || !ok {
+		t.Fatal("retired at the three-hour boundary")
+	}
+	pub.events = nil
+	update(start+int64((3*time.Hour)/time.Millisecond)+1, next)
+	if _, ok, err := m.Get(ctx, model.ProgramID(key, 6473)); err != nil || ok {
+		t.Fatalf("orphan survived the cancellation grace: found=%v err=%v", ok, err)
+	}
+	if len(pub.events) != 1 || !pub.events[0].isRemove || pub.events[0].removeID != model.ProgramID(key, 6473) {
+		t.Fatalf("notifications=%+v", pub.events)
+	}
+}
+
+func TestOrphanRetirementProtectsCurrentFutureAndKnownEvents(t *testing.T) {
+	ctx := context.Background()
+	m := newTestManager(t)
+	key := model.ServiceKey{NetworkID: 1, ServiceID: 2}
+	now := int64(100000000)
+	past := now - int64(4*time.Hour/time.Millisecond)
+	event := func(id uint16, start *int64, duration *int) model.Event {
+		return model.Event{Key: key, EventID: id, StartAt: start, DurationMS: duration}
+	}
+	current := event(1, &past, nil)
+	current.EITPresent = true
+	suspended := event(2, &past, nil)
+	suspended.EITFollowing = true
+	suspended.EITSuspended = true
+	future := event(3, testPtr(now+1000), nil)
+	unknown := event(4, nil, testPtr(1000))
+	history := event(5, &past, testPtr(1000))
+	orphan := event(6, &past, nil)
+	other := event(7, &past, nil)
+	other.Key.NetworkID = 2
+	rows := []*Program{}
+	for _, e := range []model.Event{current, suspended, future, unknown, history, orphan, other} {
+		rows = append(rows, FromEvent(e))
+	}
+	if err := m.store.UpsertAll(ctx, rows); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.UpsertPresentFollowing(ctx, model.PresentFollowingSnapshot{Service: key, ObservedAt: now, Events: []model.Event{current, suspended}}); err != nil {
+		t.Fatal(err)
+	}
+	for _, p := range rows {
+		_, ok, err := m.Get(ctx, p.ID)
+		if err != nil || ok == (p.EventID == 6) {
+			t.Fatalf("event %d found=%v err=%v", p.EventID, ok, err)
+		}
+	}
+}
+
+func TestOrphanRetirementRollsBackAndPublishesAfterCommit(t *testing.T) {
+	ctx := context.Background()
+	m := newTestManager(t)
+	pub := &fakeProgramEventPublisher{}
+	m.events = pub
+	key := model.ServiceKey{NetworkID: 1, ServiceID: 2}
+	old := FromEvent(model.Event{Key: key, EventID: 1, StartAt: testPtr[int64](1)})
+	if err := m.store.UpsertAll(ctx, []*Program{old}); err != nil {
+		t.Fatal(err)
+	}
+	write := m.store.(*sqliteStore).write
+	if _, err := write.ExecContext(ctx, `CREATE TRIGGER reject_orphan BEFORE DELETE ON programs BEGIN SELECT RAISE(ABORT, 'test failure'); END`); err != nil {
+		t.Fatal(err)
+	}
+	snapshot := model.PresentFollowingSnapshot{Service: key, ObservedAt: int64(4 * time.Hour / time.Millisecond)}
+	if err := m.UpsertPresentFollowing(ctx, snapshot); err == nil {
+		t.Fatal("delete unexpectedly succeeded")
+	}
+	m.flushEvents()
+	if _, ok, err := m.Get(ctx, old.ID); err != nil || !ok || len(pub.events) != 0 {
+		t.Fatalf("failed retirement mutated state: found=%v err=%v events=%v", ok, err, pub.events)
+	}
+	if _, err := write.ExecContext(ctx, `DROP TRIGGER reject_orphan`); err != nil {
+		t.Fatal(err)
+	}
+	if err := m.UpsertPresentFollowing(ctx, snapshot); err != nil {
+		t.Fatal(err)
+	}
+	m.flushEvents()
+	if len(pub.events) != 1 || !pub.events[0].isRemove || pub.events[0].removeID != old.ID {
+		t.Fatalf("notifications=%v", pub.events)
+	}
+}

internal/stream/presentfollowing/collector.go

@@ -3,6 +3,7 @@ package presentfollowing
 
 import (
 	"context"
+	"time"
 
 	"github.com/21S1298001/mahiron/internal/model"
 )
@@ -55,5 +56,5 @@ func (c *Collector) Observe(service model.ServiceKey, tableID, version, section,
 	}
 	all := append([]model.Event{}, t.events[0]...)
 	all = append(all, t.events[1]...)
-	return &model.PresentFollowingSnapshot{Service: service, Events: all}
+	return &model.PresentFollowingSnapshot{Service: service, Events: all, ObservedAt: time.Now().UnixMilli()}
 }

internal/stream/presentfollowing/collector_test.go

@@ -19,7 +19,7 @@ func TestCollectorWaitsForCompleteVersion(t *testing.T) {
 		t.Fatal("versions mixed at wraparound")
 	}
 	got := c.Observe(key, 0x4e, 0, 0, 1, []model.Event{next})
-	if got == nil || len(got.Events) != 2 || !got.Events[0].EITPresent || !got.Events[1].EITFollowing || got.Events[1].EventID != 1 {
+	if got == nil || got.ObservedAt <= 0 || len(got.Events) != 2 || !got.Events[0].EITPresent || !got.Events[1].EITFollowing || got.Events[1].EventID != 1 {
 		t.Fatalf("interruption snapshot=%+v", got)
 	}
 	// Re-observation must return the complete snapshot so failed writes can retry.