diff --git a/internal/hub/kalman_decision.go b/internal/hub/kalman_decision.go index d6a2179..8dfbae9 100644 --- a/internal/hub/kalman_decision.go +++ b/internal/hub/kalman_decision.go @@ -3,6 +3,7 @@ package hub import ( "context" "math" + "strings" "time" "github.com/formation-res/open-location-hub/internal/httpapi/gen" @@ -119,14 +120,16 @@ func (s *kalmanDecisionStage) normalizeTrackable(location gen.Location, trackabl decisionAt = now } - state, ok := s.state.GetKalmanTrackState(trackableID) + crs := locationCRS(location) + stateKey := kalmanTrackStateKey(trackableID, crs) + state, ok := s.state.GetKalmanTrackState(stateKey) if !ok { state = kalmanTrackState{} } - if state.CRS != "" && state.CRS != locationCRS(location) { + if state.CRS != "" && state.CRS != crs { state = kalmanTrackState{} } - state.CRS = locationCRS(location) + state.CRS = crs if state.LastDecision != nil && !state.LastDecision.At.IsZero() && decisionAt.Sub(state.LastDecision.At) > s.maxAge { state = kalmanTrackState{CRS: state.CRS} } @@ -187,10 +190,14 @@ func (s *kalmanDecisionStage) normalizeTrackable(location gen.Location, trackabl if ttl <= 0 { ttl = s.maxAge * kalmanDecisionTTLFactor } - s.state.SetKalmanTrackState(trackableID, state, ttl) + s.state.SetKalmanTrackState(stateKey, state, ttl) return out, emit, nil } +func kalmanTrackStateKey(trackableID, crs string) string { + return strings.TrimSpace(trackableID) + "\x00" + strings.TrimSpace(crs) +} + func trimKalmanSamples(samples []kalmanObservation, now time.Time, maxAge time.Duration, maxPoints int) []kalmanObservation { if len(samples) == 0 { return nil diff --git a/internal/hub/kalman_decision_test.go b/internal/hub/kalman_decision_test.go index 00650b2..cbc8d59 100644 --- a/internal/hub/kalman_decision_test.go +++ b/internal/hub/kalman_decision_test.go @@ -205,3 +205,53 @@ func TestKalmanDecisionStageAppliesEmitFrequencyOnlyToPublication(t *testing.T) t.Fatal("expected normalized state to continue updating despite throttling") } } + +func TestKalmanDecisionStageKeepsThrottleStateAcrossAlternatingCRS(t *testing.T) { + t.Parallel() + + now := time.Date(2026, 4, 27, 12, 0, 0, 0, time.UTC) + stage := newKalmanDecisionStage(NewProcessingState(func() time.Time { return now }), Config{ + LocationTTL: time.Minute, + KalmanLocationMaxPoints: 8, + KalmanLocationMaxAge: 10 * time.Second, + KalmanEmitMaxFrequencyHz: 1, + }, func() time.Time { return now }) + trackables := []string{"trackable-a"} + localCRS := "local" + wgs84CRS := "EPSG:4326" + + firstLocal := testLocationWithCoordinates(t, &localCRS, "zone-a", [2]float32{0, 0}) + firstLocal.Trackables = &trackables + firstLocal.TimestampGenerated = &now + localResults, err := stage.Process(context.Background(), firstLocal) + if err != nil { + t.Fatalf("first local process failed: %v", err) + } + if !localResults[0].Emit { + t.Fatal("expected first local sample to emit") + } + + now = now.Add(100 * time.Millisecond) + firstWGS84 := testLocationWithCoordinates(t, &wgs84CRS, "zone-a", [2]float32{8.5, 47.3}) + firstWGS84.Trackables = &trackables + firstWGS84.TimestampGenerated = &now + wgsResults, err := stage.Process(context.Background(), firstWGS84) + if err != nil { + t.Fatalf("first wgs84 process failed: %v", err) + } + if !wgsResults[0].Emit { + t.Fatal("expected first wgs84 sample to emit") + } + + now = now.Add(100 * time.Millisecond) + secondLocal := testLocationWithCoordinates(t, &localCRS, "zone-a", [2]float32{1, 0}) + secondLocal.Trackables = &trackables + secondLocal.TimestampGenerated = &now + localResults, err = stage.Process(context.Background(), secondLocal) + if err != nil { + t.Fatalf("second local process failed: %v", err) + } + if localResults[0].Emit { + t.Fatal("expected second local sample to remain publication-throttled after wgs84 input") + } +} diff --git a/internal/hub/service.go b/internal/hub/service.go index 5f47b91..1598408 100644 --- a/internal/hub/service.go +++ b/internal/hub/service.go @@ -1411,10 +1411,12 @@ func (s *Service) processNativeLocation(ctx context.Context, location gen.Locati span.End() return err } - if _, err := s.publishNativeTrackableMotions(stageCtx, location); err != nil { - span.RecordError(err) - span.End() - return err + if !s.cfg.KalmanFilterEnabled { + if _, err := s.publishNativeTrackableMotions(stageCtx, location); err != nil { + span.RecordError(err) + span.End() + return err + } } if s.derivedQueue != nil { s.derivedQueue.Submit(derivedLocationWork{Context: stageCtx, Location: location, EnqueuedAt: time.Now()}) diff --git a/internal/hub/service_test.go b/internal/hub/service_test.go index ea33f07..c7d146e 100644 --- a/internal/hub/service_test.go +++ b/internal/hub/service_test.go @@ -1629,6 +1629,53 @@ func TestProcessNativeLocationPublishesNativeAndQueuesDecisionWork(t *testing.T) } } +func TestProcessNativeLocationSkipsNativeTrackableMotionsWhenKalmanEnabled(t *testing.T) { + t.Parallel() + + bus := NewEventBus() + ch, unsubscribe := bus.Subscribe(8) + defer unsubscribe() + queue := &capturingDerivedSubmitter{} + crs := "EPSG:4326" + location := testLocationWithCoordinates(t, &crs, "external-source", [2]float32{8.5, 47.3}) + trackables := []string{"trackable-a"} + location.Trackables = &trackables + + service := &Service{ + bus: bus, + derivedQueue: queue, + cfg: Config{ + DerivedLocationBuffer: 16, + LocationTTL: time.Minute, + DedupTTL: time.Minute, + CollisionStateTTL: time.Minute, + CollisionCollidingDebounce: time.Second, + MetadataReconcileInterval: time.Second, + KalmanFilterEnabled: true, + }, + metadata: &MetadataCache{snapshot: newMetadataSnapshot(nil, nil, nil, nil)}, + state: NewProcessingState(time.Now), + logger: zapTestLogger(t), + } + + if err := service.processNativeLocation(context.Background(), location); err != nil { + t.Fatalf("processNativeLocation failed: %v", err) + } + events := collectEvents(ch, 2) + if len(events) != 1 { + t.Fatalf("expected only the native location event, got %d events", len(events)) + } + if events[0].Kind != EventLocation { + t.Fatalf("expected native location event, got %s", events[0].Kind) + } + if got := decodeEventLocation(t, events[0]); got.Source != location.Source { + t.Fatalf("unexpected native location source: %s", got.Source) + } + if len(queue.works) != 1 { + t.Fatalf("expected one queued decision work item, got %d", len(queue.works)) + } +} + func testPolicy() proximityResolutionPolicy { return proximityResolutionPolicy{ ExitGraceDuration: 15 * time.Second,