perf: bound latest-signal queries to a lookback window

e2e/setup_test.go

@@ -51,7 +51,6 @@ func GetTestServices(t *testing.T) *TestServices {
 			MaxRequestDuration:   "1m",
 			VINVCDataVersion:     "VINVCv1.0",
 			ChainID:              137,
-			DeviceLastSeenBinHrs: 3,
 		}
 
 		// Setup services

internal/config/settings.go

@@ -16,7 +16,7 @@ type Settings struct {
 	TokenExchangeIssuer          string          `yaml:"TOKEN_EXCHANGE_ISSUER_URL"`
 	VehicleNFTAddress            common.Address  `yaml:"VEHICLE_NFT_ADDRESS"`
 	MaxRequestDuration           string          `yaml:"MAX_REQUEST_DURATION"`
-	DeviceLastSeenBinHrs         int64           `yaml:"DEVICE_LAST_SEEN_BIN_HOURS"`
+	LatestSignalsLookbackDays    int64           `yaml:"LATEST_SIGNALS_LOOKBACK_DAYS"`
 	ChainID                      uint64          `yaml:"DIMO_REGISTRY_CHAIN_ID"`
 	FetchAPIGRPCEndpoint         string          `yaml:"FETCH_API_GRPC_ENDPOINT"`
 	CreditTrackerEndpoint        string          `yaml:"CREDIT_TRACKER_ENDPOINT"`

internal/repositories/repositories_test.go

@@ -20,9 +20,8 @@ import (
 )
 
 var baseSettings = config.Settings{
-	DeviceLastSeenBinHrs: 3,
-	ChainID:              80002,
-	VehicleNFTAddress:    common.HexToAddress("0x1234567890123456789012345678901234567890"),
+	ChainID:           80002,
+	VehicleNFTAddress: common.HexToAddress("0x1234567890123456789012345678901234567890"),
 }
 
 type Mocks struct {

internal/service/ch/ch.go

@@ -24,8 +24,8 @@ const (
 
 // Service is a ClickHouse service that interacts with the ClickHouse database.
 type Service struct {
-	conn              clickhouse.Conn
-	lastSeenBucketHrs int64
+	conn           clickhouse.Conn
+	latestLookback time.Duration
 }
 
 // NewService creates a new ClickHouse service.
@@ -60,7 +60,14 @@ func NewService(settings config.Settings) (*Service, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to ping clickhouse: %w", err)
 	}
-	return &Service{conn: conn, lastSeenBucketHrs: settings.DeviceLastSeenBinHrs}, nil
+	var latestLookback time.Duration
+	if settings.LatestSignalsLookbackDays > 0 {
+		latestLookback = time.Duration(settings.LatestSignalsLookbackDays) * 24 * time.Hour
+	}
+	return &Service{
+		conn:           conn,
+		latestLookback: latestLookback,
+	}, nil
 }
 
 func getMaxExecutionTime(maxRequestDuration string) (int, error) {
@@ -76,9 +83,10 @@ func getMaxExecutionTime(maxRequestDuration string) (int, error) {
 
 // GetLatestSignals returns the latest signals based on the provided arguments from the ClickHouse database.
 func (s *Service) GetLatestSignals(ctx context.Context, subject string, latestArgs *model.LatestSignalsArgs) ([]*vss.Signal, error) {
-	stmt, args := getLatestQuery(subject, latestArgs)
+	lookbackFrom := s.lookbackFrom()
+	stmt, args := getLatestQuery(subject, latestArgs, lookbackFrom)
 	if latestArgs.IncludeLastSeen {
-		lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &latestArgs.SignalArgs)
+		lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &latestArgs.SignalArgs, lookbackFrom)
 		stmt, args = unionAll([]string{stmt, lastSeenStmt}, [][]any{args, lastSeenArgs})
 	}
 
@@ -89,6 +97,15 @@ func (s *Service) GetLatestSignals(ctx context.Context, subject string, latestAr
 	return signals, nil
 }
 
+// lookbackFrom returns the earliest timestamp the "latest" queries will
+// scan, or the zero Time if no lower bound is configured.
+func (s *Service) lookbackFrom() time.Time {
+	if s.latestLookback <= 0 {
+		return time.Time{}
+	}
+	return time.Now().Add(-s.latestLookback)
+}
+
 // GetAggregatedSignals returns a slice of aggregated signals based on the provided arguments from the ClickHouse database.
 // The signals are sorted by timestamp in ascending order.
 // The timestamp on each signal is for the start of the interval.

internal/service/ch/ch_test.go

@@ -54,9 +54,8 @@ func (c *CHServiceTestSuite) SetupSuite() {
 	c.Require().NoError(err, "Failed to run migrations")
 
 	settings := config.Settings{
-		Clickhouse:           cfg,
-		MaxRequestDuration:   "1s",
-		DeviceLastSeenBinHrs: 3,
+		Clickhouse:         cfg,
+		MaxRequestDuration: "1s",
 	}
 	c.chService, err = NewService(settings)
 	c.Require().NoError(err, "Failed to create repository")

internal/service/ch/queries.go

@@ -318,7 +318,7 @@ WHERE
 GROUP BY
   name
 */
-func getLatestQuery(subject string, latestArgs *model.LatestSignalsArgs) (string, []any) {
+func getLatestQuery(subject string, latestArgs *model.LatestSignalsArgs, lookbackFrom time.Time) (string, []any) {
 	signalNames := make([]string, 0, len(latestArgs.SignalNames))
 	for name := range latestArgs.SignalNames {
 		signalNames = append(signalNames, name)
@@ -337,6 +337,11 @@ func getLatestQuery(subject string, latestArgs *model.LatestSignalsArgs) (string
 		qm.Select(latestLocation),
 		qm.From(vss.TableName),
 		qm.Where(subjectWhere, subject),
+	}
+	if !lookbackFrom.IsZero() {
+		mods = append(mods, whereTimestampFrom(lookbackFrom))
+	}
+	mods = append(mods,
 		qm.Expr(
 			qm.WhereIn(nameIn, signalNames),
 			qm.Or2(
@@ -350,7 +355,7 @@ func getLatestQuery(subject string, latestArgs *model.LatestSignalsArgs) (string
 			),
 		),
 		qm.GroupBy(vss.NameCol),
-	}
+	)
 	mods = append(mods, getFilterMods(latestArgs.Filter)...)
 	return newQuery(mods...)
 }
@@ -369,7 +374,7 @@ FROM
 WHERE
 	subject = '...'
 */
-func getLastSeenQuery(subject string, sigArgs *model.SignalArgs) (string, []any) {
+func getLastSeenQuery(subject string, sigArgs *model.SignalArgs, lookbackFrom time.Time) (string, []any) {
 	if sigArgs == nil {
 		return "", nil
 	}
@@ -382,6 +387,9 @@ func getLastSeenQuery(subject string, sigArgs *model.SignalArgs) (string, []any)
 		qm.From(vss.TableName),
 		qm.Where(subjectWhere, subject),
 	}
+	if !lookbackFrom.IsZero() {
+		mods = append(mods, whereTimestampFrom(lookbackFrom))
+	}
 	mods = append(mods, getFilterMods(sigArgs.Filter)...)
 	return newQuery(mods...)
 }

settings.sample.yaml

@@ -11,5 +11,7 @@ MANUFACTURER_NFT_ADDRESS: 0xA4ad0F9c722588910791A9BAC63ADbB365614Bc7
 PARQUET_BUCKET: dimo-iceberg-prod
 VINVC_DATA_VERSION: 'VINVCv0.0'
 IDENTITY_API_REQUEST_TIMEOUT_SECONDS: 5
-DEVICE_LAST_SEEN_BIN_HOURS: 3
 MAX_REQUEST_DURATION: 30s
+# Upper bound on how far back the "latest signal" queries scan. Set to 0
+# to disable (full scan).
+LATEST_SIGNALS_LOOKBACK_DAYS: 45