Carry the input time with each SSE event and mode change

plan/sse-data.md

@@ -153,7 +153,7 @@ than flattened.
 
 `ModeChanged` should ideally be represented by serialization of the phcsync
 observability type used on the daemon observability bus. The current callback
-is `ModeChanged(oldMode, newMode phcsync.Mode)`, so this may require a named
+is `ModeChanged(oldMode, newMode phcsync.Mode, t time.Time)`, so this may require a named
 payload type before it can replace `ModeSSE`.
 
 ### Init session events

time/app/daemon/daemon.go

@@ -32,7 +32,6 @@ import (
 	"github.com/jclark/satpulse/time/internal/ts"
 	"github.com/jclark/satpulse/time/lib/ntpshm"
 	"github.com/jclark/satpulse/time/lib/pmc"
-	"github.com/jclark/satpulse/time/lib/sse"
 	"github.com/jclark/satpulse/time/phc"
 )
 
@@ -175,10 +174,10 @@ func run(ctx context.Context, lg *slog.Logger, cancel context.CancelCauseFunc, c
 
 	pb := startBcast(ctx, lg, &wg, pCh)
 
-	var sseCh chan sse.Event
-	var eb *bcast.Bcast[sse.Event]
+	var sseCh chan sseobs.TimedEvent
+	var eb *bcast.Bcast[sseobs.TimedEvent]
 	if cfg.anyHTTP(HTTPConfig.gui) {
-		sseCh = make(chan sse.Event, 1)
+		sseCh = make(chan sseobs.TimedEvent, 1)
 		eb = startBcast(ctx, lg, &wg, sseCh)
 	}
 	// Shut down the broadcast goroutines when the context is cancelled.
@@ -430,7 +429,7 @@ func NewDispatcher(
 }
 
 // newSSEObserver creates SSE observer if any HTTP endpoint needs GUI
-func newSSEObserver(cfg *Config, sseCh chan<- sse.Event, lg *slog.Logger, gcfg *gpscfg.Result) *sseobs.SSEObserver {
+func newSSEObserver(cfg *Config, sseCh chan<- sseobs.TimedEvent, lg *slog.Logger, gcfg *gpscfg.Result) *sseobs.SSEObserver {
 	if cfg.anyHTTP(HTTPConfig.gui) {
 		return sseobs.New(sseCh, cfg.LeapSecond.leapSecond(), lg, gcfg)
 	}

time/app/daemon/http.go

@@ -15,7 +15,6 @@ import (
 	"github.com/jclark/satpulse/time/internal/proxy"
 	"github.com/jclark/satpulse/time/internal/sseobs"
 	"github.com/jclark/satpulse/time/internal/web"
-	"github.com/jclark/satpulse/time/lib/sse"
 )
 
 type HTTPConfig struct {
@@ -62,7 +61,7 @@ func registerPprofHandlers(mux *http.ServeMux) {
 	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
 }
 
-func startHTTP(ctx context.Context, lg *slog.Logger, wg *sync.WaitGroup, cfg []HTTPConfig, b *bcast.Bcast[sse.Event], sseObs *sseobs.SSEObserver, promObs *promobs.PrometheusObserver, posObs *positionObserver) error {
+func startHTTP(ctx context.Context, lg *slog.Logger, wg *sync.WaitGroup, cfg []HTTPConfig, b *bcast.Bcast[sseobs.TimedEvent], sseObs *sseobs.SSEObserver, promObs *promobs.PrometheusObserver, posObs *positionObserver) error {
 	if len(cfg) == 0 {
 		return nil
 	}
@@ -139,7 +138,7 @@ func startHTTP(ctx context.Context, lg *slog.Logger, wg *sync.WaitGroup, cfg []H
 	return nil
 }
 
-func sseHandleRequest(ctx context.Context, lg *slog.Logger, w http.ResponseWriter, r *http.Request, b *bcast.Bcast[sse.Event], sseObs *sseobs.SSEObserver) {
+func sseHandleRequest(ctx context.Context, lg *slog.Logger, w http.ResponseWriter, r *http.Request, b *bcast.Bcast[sseobs.TimedEvent], sseObs *sseobs.SSEObserver) {
 	defer lg.Debug("about to exit HTTP SSE request handler")
 	lg.Debug("starting to handle HTTP SSE request")
 	w.Header().Set("Content-Type", "text/event-stream")
@@ -174,7 +173,7 @@ func sseHandleRequest(ctx context.Context, lg *slog.Logger, w http.ResponseWrite
 				lg.Debug("broadcast channel to SSE request handler closed")
 				return
 			}
-			lg.Debug("received a broadcast SSE event", "event", event)
+			lg.Debug("received a broadcast SSE event", "event", event.Event)
 			// XXX: handle error
 			_, err := w.Write(([]byte)(event.Format()))
 			if err != nil {

time/app/daemon/http_test.go

@@ -5,7 +5,7 @@ import (
 	"log/slog"
 	"testing"
 
-	"github.com/jclark/satpulse/time/lib/sse"
+	"github.com/jclark/satpulse/time/internal/sseobs"
 )
 
 func TestConfigHTTPFeatureChecks(t *testing.T) {
@@ -103,7 +103,7 @@ func TestHTTPObserverFactoriesUseFeatureChecks(t *testing.T) {
 		Metrics:  httpBool(false),
 		Position: httpBool(false),
 	}}}
-	sseCh := make(chan sse.Event, 1)
+	sseCh := make(chan sseobs.TimedEvent, 1)
 	obs := newSSEObserver(cfg, sseCh, lg, nil)
 	if obs == nil {
 		t.Fatalf("newSSEObserver() = nil")

time/internal/gpsevent/dispatcher.go

@@ -249,7 +249,7 @@ func (d *Dispatcher) Run(tsCh <-chan ts.Event, ppsCh <-chan pps.CandidateEdge, p
 				continue
 			}
 			if e.Kind == ts.PauseEvent {
-				d.controller.Pause()
+				d.controller.Pause(e.TReadMono.Sys)
 				continue
 			}
 			if e.Kind == ts.ResumeEvent {
@@ -576,7 +576,7 @@ func (d *Dispatcher) Time(mt *gpsprot.TimeMsg, tRead time.Time) {
 
 	// Notify controller that a time message arrived
 	if d.controller != nil {
-		d.controller.TimeMessage()
+		d.controller.TimeMessage(tRead)
 	}
 
 	d.timeTicker.Time(mt, tRead)

time/internal/gpsevent/event_log_replay.go

@@ -66,7 +66,7 @@ func (rs *replaySampler) Sample(data phcsync.Sample) {
 	rs.count++
 }
 
-func (rs *replaySampler) ModeChanged(_, _ phcsync.Mode) {}
+func (rs *replaySampler) ModeChanged(_, _ phcsync.Mode, _ time.Time) {}
 
 func (rs *replaySampler) logDateTime(ref ptime.Time) string {
 	// Round because the reference time may have had a pulse offset applied

time/internal/gpsevent/replay.go

@@ -57,7 +57,7 @@ func (r *Replayer) replayEvent(bytes []byte) error {
 		})
 	case *gpsprot.TimeMsg:
 		r.timeMsgBuffer.Time(data, tRead)
-		r.ctrl.TimeMessage()
+		r.ctrl.TimeMessage(tRead)
 	case *gpsprot.LeapSecondMsg:
 		var ls ptime.LeapSecond
 		if data.UpdateLeapSecond(&ls) {

time/internal/gpsevent/replay_test.go

@@ -86,4 +86,4 @@ func (s *replaySampler) Sample(data phcsync.Sample) {
 	s.sampleCount++
 }
 
-func (s *replaySampler) ModeChanged(_, _ phcsync.Mode) {}
+func (s *replaySampler) ModeChanged(_, _ phcsync.Mode, _ time.Time) {}

time/internal/obs/observer.go

@@ -96,10 +96,10 @@ func (m *MultiObserver) Sample(data phcsync.Sample) {
 }
 
 // ModeChanged implements phcsync.Observer by type-asserting handlers to Observer
-func (m *MultiObserver) ModeChanged(oldMode, newMode phcsync.Mode) {
+func (m *MultiObserver) ModeChanged(oldMode, newMode phcsync.Mode, t time.Time) {
 	for h := range m.Handlers() {
 		if o, ok := h.(phcsync.Observer); ok {
-			o.ModeChanged(oldMode, newMode)
+			o.ModeChanged(oldMode, newMode, t)
 		}
 	}
 }
@@ -169,7 +169,7 @@ type DefaultObserver struct {
 func (o *DefaultObserver) Sample(data phcsync.Sample) {}
 
 // ModeChanged implements phcsync.Observer as a no-op
-func (o *DefaultObserver) ModeChanged(_, _ phcsync.Mode) {}
+func (o *DefaultObserver) ModeChanged(_, _ phcsync.Mode, _ time.Time) {}
 
 // Tick implements Observer as a no-op
 func (o *DefaultObserver) Tick(_ *gpsprot.TimeMsg, _ time.Time) {}

time/internal/obs/observer_test.go

@@ -26,7 +26,7 @@ func (m *mockObserver) Sample(data phcsync.Sample) {
 	m.sampleCount++
 }
 
-func (m *mockObserver) ModeChanged(_, _ phcsync.Mode) {}
+func (m *mockObserver) ModeChanged(_, _ phcsync.Mode, _ time.Time) {}
 
 func (m *mockObserver) Tick(_ *gpsprot.TimeMsg, _ time.Time) {
 	m.tickCount++

time/internal/phcsync/controller.go

@@ -217,7 +217,7 @@ func (c *Controller) SetTimeMsgBuffer(buf TimeMsgBuffer) {
 	}
 	c.timeMsgBuffer = buf
 	// Initialize to Reset mode now that we have the timeMsgBuffer
-	c.changeMode(ModeReset)
+	c.changeMode(ModeReset, time.Time{})
 }
 
 // PulseEdge handles edge timestamp events from the PHC.
@@ -228,14 +228,15 @@ func (c *Controller) PulseEdge(edge PulseEdge) {
 }
 
 // TimeMessage handles notification that a time message occurred.
-func (c *Controller) TimeMessage() {
+// tRead is when the message was read.
+func (c *Controller) TimeMessage(tRead time.Time) {
 	// The reset triggered here would be rejected by the drift-rate check against
 	// the tracking-mode persisted sample (see TrackingConfig.PersistThreshold),
 	// since a stale-firmware leap correction looks like ~1s of drift. The default
 	// PersistThreshold (900s) exceeds the 12.5-minute GPS leap-subframe worst case,
 	// so the correction is detected before any sample is persisted.
 	if c.detectLeap != nil && c.detectLeap() {
-		c.changeMode(ModeReset)
+		c.changeMode(ModeReset, tRead)
 		return
 	}
 	sample := c.sampleGen.timeMessageSample()
@@ -271,7 +272,7 @@ func (c *Controller) processSample(sample *Sample) {
 	sample.Mode = c.mode
 	c.observer.Sample(*sample)
 	if mode != c.mode {
-		c.changeMode(mode)
+		c.changeMode(mode, sample.Sys)
 	}
 }
 
@@ -308,9 +309,10 @@ func (c *Controller) LeapSecond(ls ptime.LeapSecond) {
 // Pause handles pause events from the timestamp worker.
 // This is called when the network interface loses carrier.
 // It resets sync state and transitions back to reset mode.
-func (c *Controller) Pause() {
+// t is when the carrier loss was seen.
+func (c *Controller) Pause(t time.Time) {
 	c.lg.Info("controller paused (carrier lost), transitioning to reset mode")
-	c.changeMode(ModeReset)
+	c.changeMode(ModeReset, t)
 }
 
 // Tick handles regular tick events (0.25s intervals).
@@ -369,7 +371,9 @@ func (c *Controller) Close() {
 	}
 }
 
-func (c *Controller) changeMode(mode Mode) {
+// changeMode switches to mode; t is the system time of the input that
+// caused the change, as for Observer.ModeChanged.
+func (c *Controller) changeMode(mode Mode, t time.Time) {
 	if c.mode == mode {
 		return
 	}
@@ -418,7 +422,7 @@ func (c *Controller) changeMode(mode Mode) {
 	c.mode = mode
 
 	c.gmUpdate()
-	c.observer.ModeChanged(old, mode)
+	c.observer.ModeChanged(old, mode, t)
 }
 
 // getDetectLeap builds the parameter-less leap-detection closure to

time/internal/phcsync/controller_test.go

@@ -186,12 +186,13 @@ type recordingSampler struct {
 
 type modeChange struct {
 	old, new Mode
+	t        time.Time
 }
 
 func (s *recordingSampler) Sample(data Sample) { s.samples = append(s.samples, data) }
 
-func (s *recordingSampler) ModeChanged(old, new Mode) {
-	s.modeChanges = append(s.modeChanges, modeChange{old, new})
+func (s *recordingSampler) ModeChanged(old, new Mode, t time.Time) {
+	s.modeChanges = append(s.modeChanges, modeChange{old, new, t})
 }
 
 // fakeClock is a minimal phcsync.Clock implementation for tests.
@@ -232,7 +233,27 @@ func forceConverging(t *testing.T, c *Controller, rsgDetectLeap func(time.Durati
 	}
 	rsg.detectLeap = rsgDetectLeap
 	c.lastSample = &Sample{Kind: SampleOK}
-	c.changeMode(ModeConverging)
+	c.changeMode(ModeConverging, time.Time{})
+}
+
+// modeProc is a sampleProcessor that always asks for mode.
+type modeProc struct{ mode Mode }
+
+func (p modeProc) processSample(*Sample) (phcAction, Mode) { return phcAction{}, p.mode }
+
+// TestSampleModeChangeTime checks that a mode change caused by a sample
+// is reported with the sample's Sys.
+func TestSampleModeChangeTime(t *testing.T) {
+	c := newTestController(t, &fakeBuffer{})
+	forceConverging(t, c, func(time.Duration) bool { return false })
+	c.sampleProc = modeProc{ModeReset}
+	sys := time.Unix(4, 0)
+	c.processPresentSample(&Sample{Kind: SampleOK, Sys: sys})
+	rec := c.observer.(*recordingSampler)
+	want := modeChange{ModeConverging, ModeReset, sys}
+	if got := rec.modeChanges[len(rec.modeChanges)-1]; got != want {
+		t.Errorf("last ModeChanged = %v, want %v", got, want)
+	}
 }
 
 // TestTimeMessageDetectsLeap exercises the Controller.TimeMessage
@@ -283,10 +304,18 @@ func TestTimeMessageDetectsLeap(t *testing.T) {
 			} else {
 				c.detectLeap = nil
 			}
-			c.TimeMessage()
+			tRead := time.Unix(5, 0)
+			c.TimeMessage(tRead)
 			if c.Mode() != tt.expectMode {
 				t.Errorf("Mode() = %v, want %v", c.Mode(), tt.expectMode)
 			}
+			if tt.expectMode == ModeReset {
+				rec := c.observer.(*recordingSampler)
+				want := modeChange{ModeConverging, ModeReset, tRead}
+				if got := rec.modeChanges[len(rec.modeChanges)-1]; got != want {
+					t.Errorf("last ModeChanged = %v, want %v", got, want)
+				}
+			}
 			if called != tt.expectInvoked {
 				t.Errorf("detectLeap invoked = %v, want %v", called, tt.expectInvoked)
 			}
@@ -328,7 +357,7 @@ func TestChangeModeWiringDetectLeap(t *testing.T) {
 		if c.detectLeap == nil {
 			t.Fatalf("preconditions: closure should be set after leaving reset")
 		}
-		c.changeMode(ModeReset)
+		c.changeMode(ModeReset, time.Time{})
 		if c.detectLeap != nil {
 			t.Errorf("expected detectLeap to be cleared on entering reset, got non-nil")
 		}
@@ -371,16 +400,18 @@ func TestModeChangedFiresWithoutSample(t *testing.T) {
 	c.SetTimeMsgBuffer(&fakeBuffer{})
 	// Drive sampleless transitions through the controller's internal API.
 	c.lastSample = &Sample{Kind: SampleOK}
-	c.changeMode(ModeConverging)
-	c.changeMode(ModeTracking)
+	t1, t2, t3 := time.Unix(1, 0), time.Unix(2, 0), time.Unix(3, 0)
+	c.changeMode(ModeConverging, t1)
+	c.changeMode(ModeTracking, t2)
 	// The case Codex flagged: carrier loss while tracking.
-	c.Pause()
+	c.Pause(t3)
 
+	// The initial transition has no input, so its time is zero.
 	want := []modeChange{
-		{ModeInvalid, ModeReset},
-		{ModeReset, ModeConverging},
-		{ModeConverging, ModeTracking},
-		{ModeTracking, ModeReset},
+		{ModeInvalid, ModeReset, time.Time{}},
+		{ModeReset, ModeConverging, t1},
+		{ModeConverging, ModeTracking, t2},
+		{ModeTracking, ModeReset, t3},
 	}
 	if len(rec.modeChanges) != len(want) {
 		t.Fatalf("ModeChanged calls: got %d (%v), want %d (%v)",

time/internal/phcsync/grandmaster_test.go

@@ -153,7 +153,7 @@ func TestControllerGrandmasterClose(t *testing.T) {
 			f.request(t).complete(nil)
 			f.tick()
 
-			f.c.Pause()
+			f.c.Pause(time.Now())
 			pending := f.request(t)
 			checkGrandmasterSettings(t, pending.settings, 37, 0, false)
 			f.close()
@@ -281,7 +281,7 @@ func (f *grandmasterController) step() {
 			GNSS:    gpsprot.GPS, Ref: gpsprot.PostPulse,
 			Tag: gpsreg.TagUBX, NativeMsgID: "NAV-PVT",
 		}, sys)
-		f.c.TimeMessage()
+		f.c.TimeMessage(sys)
 	}
 	f.tick()
 	synctest.Wait()

time/internal/phcsync/sampler.go

@@ -53,7 +53,14 @@ type Observer interface {
 	// by a sample (e.g. carrier-loss Pause). At controller startup the
 	// initial transition out of ModeInvalid into ModeReset is also
 	// reported here.
-	ModeChanged(oldMode, newMode Mode)
+	//
+	// t is the system time of the event behind the transition: for a
+	// transition caused by a sample, the sample's Sys, which is the
+	// estimated time of the pulse and can be up to about a second before
+	// the input that produced the sample was read; for leap-second
+	// detection, the read time of the time message; for carrier loss,
+	// when it was seen. It is zero for the initial transition at startup.
+	ModeChanged(oldMode, newMode Mode, t time.Time)
 }
 
 // MultiObserver fans out events to multiple observers.
@@ -74,9 +81,9 @@ func (m *MultiObserver) Sample(data Sample) {
 }
 
 // ModeChanged implements Observer by calling ModeChanged on all observers.
-func (m *MultiObserver) ModeChanged(oldMode, newMode Mode) {
+func (m *MultiObserver) ModeChanged(oldMode, newMode Mode, t time.Time) {
 	for _, o := range m.observers {
-		o.ModeChanged(oldMode, newMode)
+		o.ModeChanged(oldMode, newMode, t)
 	}
 }
 

time/internal/promobs/prometheus.go

@@ -285,7 +285,7 @@ func makeBuckets(clockAccuracyNanos float64) []float64 {
 // ModeChanged drives sync_mode and sync_status from the new controller
 // mode. Both gauges therefore reflect current state — including transitions
 // that aren't sample-driven, such as a Pause() on carrier loss.
-func (p *PrometheusObserver) ModeChanged(_, newMode phcsync.Mode) {
+func (p *PrometheusObserver) ModeChanged(_, newMode phcsync.Mode, _ time.Time) {
 	p.syncModeGauge.Set(float64(newMode))
 	if newMode.InSync() {
 		p.inSyncGauge.Set(1)