From e210fb09f9cd9f0b355c846743e57ee61fa4a290 Mon Sep 17 00:00:00 2001 From: James Clark Date: Sat, 24 Oct 2026 13:46:08 +0700 Subject: [PATCH 1/2] Carry the input time with each SSE event and mode change The fix for #497 will bound how far an SSE client can fall behind by giving each write a deadline counted from when the data it carries was produced. Neither the SSE events nor the PHC mode-change notifications carried such a time. Observer.ModeChanged now takes the system time of the event behind the transition: the sample's Sys (the estimated time of the pulse), the read time of the time message that triggered leap detection, or when carrier loss was seen, which the ts pause event now records. It is zero for the initial transition at startup. Controller.TimeMessage and Controller.Pause take those times from their callers, so phcsync still never reads the clock itself. sseobs sends a TimedEvent, an sse.Event with that time, on the event bcast: the read time of the GPS message, the sample's Sys, or the mode change's time. The SSE handler still writes the event as before and ignores the time, so there is no change in behaviour. --- plan/sse-data.md | 2 +- time/app/daemon/daemon.go | 9 ++- time/app/daemon/http.go | 7 +- time/app/daemon/http_test.go | 4 +- time/internal/gpsevent/dispatcher.go | 4 +- time/internal/gpsevent/event_log_replay.go | 2 +- time/internal/gpsevent/replay.go | 2 +- time/internal/gpsevent/replay_test.go | 2 +- time/internal/obs/observer.go | 6 +- time/internal/obs/observer_test.go | 2 +- time/internal/phcsync/controller.go | 20 +++--- time/internal/phcsync/controller_test.go | 55 ++++++++++++---- time/internal/phcsync/grandmaster_test.go | 4 +- time/internal/phcsync/sampler.go | 13 +++- time/internal/promobs/prometheus.go | 2 +- .../promobs/prometheus_replay_test.go | 6 +- time/internal/sseobs/sse.go | 41 +++++++----- time/internal/sseobs/sse_test.go | 65 ++++++++++++------- time/internal/syncsim/syncsim.go | 6 +- time/internal/ts/clock.go | 4 +- 20 files changed, 161 insertions(+), 95 deletions(-) diff --git a/plan/sse-data.md b/plan/sse-data.md index b3f0ffc80..863f7cc98 100644 --- a/plan/sse-data.md +++ b/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 diff --git a/time/app/daemon/daemon.go b/time/app/daemon/daemon.go index 3ced576d8..ab7dbde49 100644 --- a/time/app/daemon/daemon.go +++ b/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) } diff --git a/time/app/daemon/http.go b/time/app/daemon/http.go index 2bed8de1b..368005ae9 100644 --- a/time/app/daemon/http.go +++ b/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 { diff --git a/time/app/daemon/http_test.go b/time/app/daemon/http_test.go index 6cd5d7615..244cc711d 100644 --- a/time/app/daemon/http_test.go +++ b/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") diff --git a/time/internal/gpsevent/dispatcher.go b/time/internal/gpsevent/dispatcher.go index b92401090..3389a4f4d 100644 --- a/time/internal/gpsevent/dispatcher.go +++ b/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) diff --git a/time/internal/gpsevent/event_log_replay.go b/time/internal/gpsevent/event_log_replay.go index 135c44dfe..80833d7cc 100644 --- a/time/internal/gpsevent/event_log_replay.go +++ b/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 diff --git a/time/internal/gpsevent/replay.go b/time/internal/gpsevent/replay.go index 3877ebac7..baa8639ff 100644 --- a/time/internal/gpsevent/replay.go +++ b/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) { diff --git a/time/internal/gpsevent/replay_test.go b/time/internal/gpsevent/replay_test.go index e1f4dedc7..8ae0620b8 100644 --- a/time/internal/gpsevent/replay_test.go +++ b/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) {} diff --git a/time/internal/obs/observer.go b/time/internal/obs/observer.go index 02fb5c4e9..03098a86a 100644 --- a/time/internal/obs/observer.go +++ b/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) {} diff --git a/time/internal/obs/observer_test.go b/time/internal/obs/observer_test.go index 1650f64e5..d084c8246 100644 --- a/time/internal/obs/observer_test.go +++ b/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++ diff --git a/time/internal/phcsync/controller.go b/time/internal/phcsync/controller.go index c296d3028..3175ff9d8 100644 --- a/time/internal/phcsync/controller.go +++ b/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 diff --git a/time/internal/phcsync/controller_test.go b/time/internal/phcsync/controller_test.go index c758ec5c8..03ab40b35 100644 --- a/time/internal/phcsync/controller_test.go +++ b/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,13 +233,33 @@ 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{}) } // TestTimeMessageDetectsLeap exercises the Controller.TimeMessage // gating around the captured detectLeap closure. Kill-switch and // capture-vs-not-capture wiring is covered separately by // TestChangeModeWiringDetectLeap. +// 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) + } +} + func TestTimeMessageDetectsLeap(t *testing.T) { tests := []struct { name string @@ -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)", diff --git a/time/internal/phcsync/grandmaster_test.go b/time/internal/phcsync/grandmaster_test.go index 4e50454ae..41a1879d4 100644 --- a/time/internal/phcsync/grandmaster_test.go +++ b/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() diff --git a/time/internal/phcsync/sampler.go b/time/internal/phcsync/sampler.go index 4002e488e..6d7722c68 100644 --- a/time/internal/phcsync/sampler.go +++ b/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) } } diff --git a/time/internal/promobs/prometheus.go b/time/internal/promobs/prometheus.go index e743c3904..8886d57ff 100644 --- a/time/internal/promobs/prometheus.go +++ b/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) diff --git a/time/internal/promobs/prometheus_replay_test.go b/time/internal/promobs/prometheus_replay_test.go index 8b545e2b9..46682101f 100644 --- a/time/internal/promobs/prometheus_replay_test.go +++ b/time/internal/promobs/prometheus_replay_test.go @@ -63,10 +63,10 @@ func TestReplay(t *testing.T) { t.Logf("replayed %d NavEpoch, %d Satellites events", nNavEpoch, nSatellites) // Also exercise Sample() and ModeChanged() with hand-crafted PHC samples. - obs.ModeChanged(phcsync.ModeInvalid, phcsync.ModeReset) + obs.ModeChanged(phcsync.ModeInvalid, phcsync.ModeReset, time.Time{}) obs.Sample(phcsync.Sample{Kind: phcsync.SampleMissing, Mode: phcsync.ModeReset, Freq: 1000}) obs.Sample(phcsync.Sample{Kind: phcsync.SampleMissing, Mode: phcsync.ModeReset, Freq: 2000}) - obs.ModeChanged(phcsync.ModeReset, phcsync.ModeTracking) + obs.ModeChanged(phcsync.ModeReset, phcsync.ModeTracking, time.Time{}) obs.Sample(phcsync.Sample{ Kind: phcsync.SampleOK, Mode: phcsync.ModeTracking, Freq: 500, Offset: 50 * time.Nanosecond, FreqDelta: 0.1, @@ -188,7 +188,7 @@ doneSignalCheck: // A ModeChanged notification — without any accompanying sample — must // drive both gauges immediately, including paths like carrier-loss // Pause where the controller transitions without producing a sample. - obs.ModeChanged(phcsync.ModeTracking, phcsync.ModeReset) + obs.ModeChanged(phcsync.ModeTracking, phcsync.ModeReset, time.Time{}) mfAfterTransition := scrape(t, obs) assertGauge(t, mfAfterTransition, "satpulse_phc_sync_mode", float64(phcsync.ModeReset)) assertGauge(t, mfAfterTransition, "satpulse_phc_sync_status", 0) diff --git a/time/internal/sseobs/sse.go b/time/internal/sseobs/sse.go index 01c129fe3..11dbddb00 100644 --- a/time/internal/sseobs/sse.go +++ b/time/internal/sseobs/sse.go @@ -139,10 +139,20 @@ type CorReportSSE struct { // config changes recover within a sensible window. const corReportSourceTimeout = 30 * time.Second +// TimedEvent is an SSE event together with the system time of what it +// reports: when the GPS message was read, the estimated time of the +// pulse for a PHC sample, or the time ModeChanged gives for a mode +// change. T is zero when there is no such time, and the event should +// then be treated as made when it is received. +type TimedEvent struct { + sse.Event + T time.Time +} + // SSEObserver implements obs.Observer for Server-Sent Events type SSEObserver struct { obs.DefaultObserver - sseCh chan<- sse.Event + sseCh chan<- TimedEvent lg *slog.Logger ls ptime.LeapSecond initEvent sse.Event @@ -160,7 +170,7 @@ type SSEObserver struct { // New creates a new SSE observer with the provided channel and leap second. // The sseCh parameter must be non-nil. -func New(sseCh chan<- sse.Event, ls ptime.LeapSecond, lg *slog.Logger, cfgResult *gpscfg.Result) *SSEObserver { +func New(sseCh chan<- TimedEvent, ls ptime.LeapSecond, lg *slog.Logger, cfgResult *gpscfg.Result) *SSEObserver { if sseCh == nil { panic("sseCh must be non-nil") } @@ -197,14 +207,14 @@ func (o *SSEObserver) InitEvents() []sse.Event { // ModeChanged implements phcsync.Observer. Stores the latest mode event // for delivery to new clients and broadcasts it to current subscribers. -func (o *SSEObserver) ModeChanged(_, newMode phcsync.Mode) { +func (o *SSEObserver) ModeChanged(_, newMode phcsync.Mode, t time.Time) { ev, err := sse.Make("mode", ModeSSE{Mode: newMode.String()}) if err != nil { o.lg.Error("failed to create SSE event", "name", "mode", "err", err) return } o.modeEvent.Store(&ev) - o.sseCh <- ev + o.sseCh <- TimedEvent{ev, t} } // Release implements Observer - closes the SSE channel @@ -223,12 +233,12 @@ func (o *SSEObserver) Sample(data phcsync.Sample) { Outlier: data.Kind == phcsync.SampleOutlier, Mode: data.Mode.String(), } - o.sendSSE("phc", event) + o.sendSSE("phc", event, data.Sys) } // Tick receives a filled TimeMsg from the Dispatcher's TimeTicker. // Only emits an SSE event when the time is an integral second. -func (o *SSEObserver) Tick(mt *gpsprot.TimeMsg, _ time.Time) { +func (o *SSEObserver) Tick(mt *gpsprot.TimeMsg, tRead time.Time) { tai := mt.TAITime if tai.IsZero() { return @@ -239,7 +249,7 @@ func (o *SSEObserver) Tick(mt *gpsprot.TimeMsg, _ time.Time) { o.sendSSE("time", TimeSSE{ UTC: o.ls.FormatTime(tai), TAI: int64(tai) / 1e9, - }) + }, tRead) } // Survey implements gpsprot.MsgHandler - generates survey SSE events (copied exactly from dispatcher.go) @@ -269,12 +279,12 @@ func (o *SSEObserver) Survey(m *gpsprot.SurveyMsg, tRead time.Time) { event.Alt.Set(gpsprot.Meters(llh.Height)) } } - o.sendSSE("survey", event) + o.sendSSE("survey", event, tRead) } // Satellites implements gpsprot.MsgHandler - generates satellites SSE events func (o *SSEObserver) Satellites(msg *gpsprot.SatellitesMsg, tRead time.Time) { - o.sendSSE("satellites", SatellitesSSE{SVs: msg.SVs}) + o.sendSSE("satellites", SatellitesSSE{SVs: msg.SVs}, tRead) } // LeapSecond updates the local leap second used for time formatting. @@ -310,26 +320,27 @@ func (o *SSEObserver) CorReport(msg *gpsprot.CorReportMsg, tRead time.Time) { ChecksumOK: msg.ChecksumOK, Used: msg.Used, RTCMRefBaseID: msg.RTCMRefBaseID, - }) + }, tRead) } // NavEpochPV emits posvel and quality SSE events from the accumulated bundle. func (o *SSEObserver) NavEpochPV(msg *gpsprot.NavEpochMsg, pv *gpsprot.PVMsgBundle, tRead time.Time) { if pv := buildPosVelSSE(pv); pv != nil { - o.sendSSE("posvel", pv) + o.sendSSE("posvel", pv, tRead) } if q := buildQualitySSE(msg); q != nil { - o.sendSSE("quality", q) + o.sendSSE("quality", q, tRead) } } -// sendSSE is a helper method to send SSE events -func (o *SSEObserver) sendSSE(name string, data any) { +// sendSSE is a helper method to send SSE events; t is the system time of +// the input the event reports. +func (o *SSEObserver) sendSSE(name string, data any, t time.Time) { event, err := sse.Make(name, data) if err != nil { o.lg.Error("failed to create SSE event", "name", name, "err", err) } else { - o.sseCh <- event + o.sseCh <- TimedEvent{event, t} } } diff --git a/time/internal/sseobs/sse_test.go b/time/internal/sseobs/sse_test.go index 5b5343618..733299943 100644 --- a/time/internal/sseobs/sse_test.go +++ b/time/internal/sseobs/sse_test.go @@ -23,39 +23,42 @@ const oneYearSecs = 365 * 24 * 3600 func TestSSEObserver_Events(t *testing.T) { tests := []struct { name string - action func(*SSEObserver) + action func(obs *SSEObserver, tRead time.Time) + skipTime bool // the test sends the event itself, so its time means nothing eventType string expectedJSON string }{ { name: "init", - action: func(obs *SSEObserver) { + action: func(obs *SSEObserver, tRead time.Time) { events := obs.InitEvents() if len(events) != 1 { panic("expected single init event before any ModeChanged") } - obs.sseCh <- events[0] + obs.sseCh <- TimedEvent{Event: events[0]} }, + skipTime: true, eventType: "init", expectedJSON: `{"receiver":{"vendor":"u-blox","firmware":"HPG 1.32 PROTVER 27.31","hardware":"ZED-F9P","supportedGNSS":["GPS","GLO"]}}`, }, { name: "mode_changed", - action: func(obs *SSEObserver) { - obs.ModeChanged(phcsync.ModeReset, phcsync.ModeTracking) + action: func(obs *SSEObserver, tRead time.Time) { + obs.ModeChanged(phcsync.ModeReset, phcsync.ModeTracking, tRead) }, eventType: "mode", expectedJSON: `{"mode":"tracking"}`, }, { name: "sample_ok", - action: func(obs *SSEObserver) { + action: func(obs *SSEObserver, tRead time.Time) { obs.Sample(phcsync.Sample{ Kind: phcsync.SampleOK, Offset: 123 * time.Nanosecond, Freq: 1.5, Mode: phcsync.ModeTracking, Era: phctime.Era(0), + Sys: tRead, }) }, eventType: "phc", @@ -63,13 +66,14 @@ func TestSSEObserver_Events(t *testing.T) { }, { name: "sample_outlier", - action: func(obs *SSEObserver) { + action: func(obs *SSEObserver, tRead time.Time) { obs.Sample(phcsync.Sample{ Kind: phcsync.SampleOutlier, Offset: -456 * time.Nanosecond, Freq: -2.3, Mode: phcsync.ModeReset, Era: phctime.Era(0), + Sys: tRead, }) }, eventType: "phc", @@ -77,18 +81,18 @@ func TestSSEObserver_Events(t *testing.T) { }, { name: "time", - action: func(obs *SSEObserver) { + action: func(obs *SSEObserver, tRead time.Time) { obs.Tick(&gpsprot.TimeMsg{ Ref: gpsprot.PostPulse, TAITime: ptime.Time(oneYearSecs * 1e9), - }, time.Now()) + }, tRead) }, eventType: "time", expectedJSON: fmt.Sprintf(`{"utc":"1971-01-01T00:00:00Z","tai":%d}`, oneYearSecs), }, { name: "survey", - action: func(obs *SSEObserver) { + action: func(obs *SSEObserver, tRead time.Time) { obs.Survey(&gpsprot.SurveyMsg{ Position: gpsprot.Point3D{gpsprot.Meters(1000), gpsprot.Meters(2000), gpsprot.Meters(3000)}, Accuracy: gpsprot.Meters(5.5), @@ -96,28 +100,28 @@ func TestSSEObserver_Events(t *testing.T) { ObsCount: 150, InProgress: true, Valid: true, - }, time.Now()) + }, tRead) }, eventType: "survey", expectedJSON: `{"x":1000,"y":2000,"z":3000,"accuracy":5.5,"obsTime":30,"obsCount":150,"inProgress":true,"valid":true}`, }, { name: "survey_no_position", - action: func(obs *SSEObserver) { + action: func(obs *SSEObserver, tRead time.Time) { obs.Survey(&gpsprot.SurveyMsg{ Accuracy: gpsprot.Meters(5.5), ObsTime: 60 * gpsprot.Second, ObsCount: 60, InProgress: true, Valid: false, - }, time.Now()) + }, tRead) }, eventType: "survey", expectedJSON: `{"accuracy":5.5,"obsTime":60,"obsCount":60,"inProgress":true,"valid":false}`, }, { name: "satellites", - action: func(obs *SSEObserver) { + action: func(obs *SSEObserver, tRead time.Time) { obs.Satellites(&gpsprot.SatellitesMsg{ SVs: []gpsprot.SVInfo{ { @@ -130,7 +134,7 @@ func TestSSEObserver_Events(t *testing.T) { Used: true, }, }, - }, time.Now()) + }, tRead) }, eventType: "satellites", expectedJSON: `{"svs":[{"id":"G01","lookAngles":{"azimuth":45,"elevation":30},"signals":[{"id":"L1","cn0":0}],"used":true}]}`, @@ -139,7 +143,7 @@ func TestSSEObserver_Events(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ch := make(chan sse.Event, 1) + ch := make(chan TimedEvent, 1) cfgResult := &gpscfg.Result{ ReceiverInfo: &gpsprot.ReceiverInfo{ Vendor: "u-blox", @@ -150,9 +154,13 @@ func TestSSEObserver_Events(t *testing.T) { } obs := New(ch, ptime.LeapSecond{}, slog.Default(), cfgResult) - tt.action(obs) + tRead := time.Unix(1000, 0) + tt.action(obs, tRead) event := <-ch + if !tt.skipTime && !event.T.Equal(tRead) { + t.Errorf("event time %v, want %v", event.T, tRead) + } wireFormat := event.Format() // Check event type @@ -376,7 +384,7 @@ func TestBuildQualitySSE(t *testing.T) { } func TestNavEpochPVSSE(t *testing.T) { - ch := make(chan sse.Event, 4) + ch := make(chan TimedEvent, 4) cfgResult := &gpscfg.Result{ ReceiverInfo: &gpsprot.ReceiverInfo{Vendor: "test"}, } @@ -388,32 +396,39 @@ func TestNavEpochPVSSE(t *testing.T) { }), } // Fire NavEpochPV + tRead := time.Unix(1000, 0) obs.NavEpochPV(&gpsprot.NavEpochMsg{ FixLevel: gpsprot.FixLevelCode, SolutionDim: gpsprot.SolutionDim3D, - }, pv, time.Now()) + }, pv, tRead) // Should produce posvel then quality posvel := <-ch if !strings.Contains(posvel.Format(), "event: posvel\n") { t.Errorf("expected posvel event, got: %s", posvel.Format()) } + if !posvel.T.Equal(tRead) { + t.Errorf("posvel time %v, want %v", posvel.T, tRead) + } quality := <-ch if !strings.Contains(quality.Format(), "event: quality\n") { t.Errorf("expected quality event, got: %s", quality.Format()) } + if !quality.T.Equal(tRead) { + t.Errorf("quality time %v, want %v", quality.T, tRead) + } } // TestInitEventsAfterModeChanged checks that once ModeChanged has been // observed, InitEvents includes the cached mode event so a fresh client // learns the current mode at connect time. func TestInitEventsAfterModeChanged(t *testing.T) { - ch := make(chan sse.Event, 1) + ch := make(chan TimedEvent, 1) cfgResult := &gpscfg.Result{ReceiverInfo: &gpsprot.ReceiverInfo{Vendor: "test"}} obs := New(ch, ptime.LeapSecond{}, slog.Default(), cfgResult) if got := obs.InitEvents(); len(got) != 1 { t.Fatalf("expected 1 init event before ModeChanged, got %d", len(got)) } - obs.ModeChanged(phcsync.ModeInvalid, phcsync.ModeReset) + obs.ModeChanged(phcsync.ModeInvalid, phcsync.ModeReset, time.Time{}) <-ch // drain the broadcast emit events := obs.InitEvents() if len(events) != 2 { @@ -430,7 +445,7 @@ func TestInitEventsAfterModeChanged(t *testing.T) { t.Errorf("mode event payload should be {\"mode\":\"reset\"}, got: %s", wire) } // A subsequent transition replaces the cached mode. - obs.ModeChanged(phcsync.ModeReset, phcsync.ModeTracking) + obs.ModeChanged(phcsync.ModeReset, phcsync.ModeTracking, time.Time{}) <-ch events = obs.InitEvents() if len(events) != 2 { @@ -507,7 +522,7 @@ func TestSSEObserverCorReportLatch(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - ch := make(chan sse.Event, len(tc.events)) + ch := make(chan TimedEvent, len(tc.events)) obs := New(ch, ptime.LeapSecond{}, slog.Default(), &gpscfg.Result{}) for _, ev := range tc.events { obs.CorReport(&gpsprot.CorReportMsg{ @@ -519,7 +534,7 @@ func TestSSEObserverCorReportLatch(t *testing.T) { close(ch) var got []string for ev := range ch { - got = append(got, parseCorReportSource(t, ev)) + got = append(got, parseCorReportSource(t, ev.Event)) } if !reflect.DeepEqual(got, tc.expect) { t.Errorf("got %v\nwant %v", got, tc.expect) @@ -552,7 +567,7 @@ func parseCorReportSource(t *testing.T, ev sse.Event) string { // when GPS detection fails but the daemon continues running because no // PHC is configured. func TestNewNilCfgResult(t *testing.T) { - ch := make(chan sse.Event, 1) + ch := make(chan TimedEvent, 1) obs := New(ch, ptime.LeapSecond{}, slog.Default(), nil) if got := obs.InitEvents(); len(got) != 0 { t.Fatalf("expected 0 init events with nil cfgResult, got %d", len(got)) diff --git a/time/internal/syncsim/syncsim.go b/time/internal/syncsim/syncsim.go index e006ace6f..312252aa6 100644 --- a/time/internal/syncsim/syncsim.go +++ b/time/internal/syncsim/syncsim.go @@ -387,7 +387,7 @@ func Simulate(observers []obs.Observer, cfg Config, tsLog io.Writer, curTime *ti } msgTRead := time.Unix(0, 0).Add(time.Duration(event.Time * 1e9)) timeMsgBuf.Time(timeMsg, msgTRead) - ctrl.TimeMessage() + ctrl.TimeMessage(msgTRead) lg.Debug("delivering PrePulse message", "second", int(data.PPS), "taiTime", tMsg, @@ -417,7 +417,7 @@ func Simulate(observers []obs.Observer, cfg Config, tsLog io.Writer, curTime *ti } msgTRead := time.Unix(0, 0).Add(time.Duration(event.Time * 1e9)) timeMsgBuf.Time(timeMsg, msgTRead) - ctrl.TimeMessage() + ctrl.TimeMessage(msgTRead) lg.Debug("delivering PostPulse message", "second", int(data.PPS), "taiTime", tMsg, @@ -684,7 +684,7 @@ func handleNavSolutionMsgEvent( "second", int(data.PPS), "taiTime", tMsg, "msgTRead", msgTRead) - ctrl.TimeMessage() + ctrl.TimeMessage(msgTRead) } // handleTickEvent processes a tick event diff --git a/time/internal/ts/clock.go b/time/internal/ts/clock.go index bbcd7f835..cfd676d7b 100644 --- a/time/internal/ts/clock.go +++ b/time/internal/ts/clock.go @@ -37,7 +37,7 @@ type EventKind int const ( _ EventKind = iota EdgeEvent - PauseEvent + PauseEvent // only TReadMono.Sys is set: when carrier loss was seen ResumeEvent ) @@ -237,7 +237,7 @@ func noCarrier(flags net.Flags) bool { } func (clk *Clock) handleNoCarrier(ctx context.Context, lg *slog.Logger, tsCh chan<- Event, _ net.Flags) <-chan net.Flags { - tsCh <- Event{Kind: PauseEvent} + tsCh <- Event{Kind: PauseEvent, TReadMono: phctime.Sample{Sys: time.Now()}} lg.Info("carrier lost; waiting for it to return", "interface", clk.ifName) clk.logExttsEnable(lg, false) wCh := clk.w.SetCond(hasCarrier) From d10bd547389f07ce8c4a619bff08759b69b1a5a4 Mon Sep 17 00:00:00 2001 From: James Clark Date: Sat, 24 Oct 2026 17:13:44 +0700 Subject: [PATCH 2/2] phcsync: move TestTimeMessageDetectsLeap's doc comment back to it modeProc and TestSampleModeChangeTime were inserted between TestTimeMessageDetectsLeap and its doc comment, so the comment documented modeProc instead and the test had none. --- time/internal/phcsync/controller_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/time/internal/phcsync/controller_test.go b/time/internal/phcsync/controller_test.go index 03ab40b35..114c6c0bd 100644 --- a/time/internal/phcsync/controller_test.go +++ b/time/internal/phcsync/controller_test.go @@ -236,10 +236,6 @@ func forceConverging(t *testing.T, c *Controller, rsgDetectLeap func(time.Durati c.changeMode(ModeConverging, time.Time{}) } -// TestTimeMessageDetectsLeap exercises the Controller.TimeMessage -// gating around the captured detectLeap closure. Kill-switch and -// capture-vs-not-capture wiring is covered separately by -// TestChangeModeWiringDetectLeap. // modeProc is a sampleProcessor that always asks for mode. type modeProc struct{ mode Mode } @@ -260,6 +256,10 @@ func TestSampleModeChangeTime(t *testing.T) { } } +// TestTimeMessageDetectsLeap exercises the Controller.TimeMessage +// gating around the captured detectLeap closure. Kill-switch and +// capture-vs-not-capture wiring is covered separately by +// TestChangeModeWiringDetectLeap. func TestTimeMessageDetectsLeap(t *testing.T) { tests := []struct { name string