Throttle Kalman trackable motion streams

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

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")
+	}
+}

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()})

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,