diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index cd7c524a..2d521410 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -71,6 +71,7 @@ - `jt808_registration`：808 注册、鉴权、最新活跃和 VIN 匹配状态。 - 状态：`vehicle_identity_binding` 使用 VIN 主键；`jt808_registration` 使用 phone 主键；两张表都不保留代理自增主键和 `created_at`。 - 查询：`vehicle_identity_binding` 的 phone/device_id/plate 都是唯一键，接入侧按唯一键直查 VIN，不做无意义排序。 + - 写入链路：gateway 对 binding 查 VIN 使用短 TTL 内存缓存，包含未命中缓存，避免无 VIN 高频位置帧每条都打 MySQL。 - 生产：RDS 已在上线前删除旧 `vehicle_identity_binding_registration`、`vehicle_identity_bindings`，gateway 启动会自动创建最小 schema。 ## 字段提升规则 diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index 2061d539..6982277a 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -146,6 +146,7 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R table := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding") resolver := identity.NewMySQLResolverWithOptions(db, table, identity.MySQLResolverOptions{ LocationTouchInterval: time.Duration(envInt("JT808_REGISTRATION_LOCATION_TOUCH_INTERVAL_SECONDS", 600)) * time.Second, + LookupCacheTTL: time.Duration(envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600)) * time.Second, }) if envBool("IDENTITY_MYSQL_ENSURE_SCHEMA", true) { if err := resolver.EnsureSchema(ctx); err != nil { @@ -153,7 +154,7 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R return nil, nil, err } } - logger.Info("identity mysql resolver enabled", "table", table) + logger.Info("identity mysql resolver enabled", "table", table, "lookup_cache_ttl_seconds", envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600)) return resolver, func() { _ = db.Close() }, nil } diff --git a/go/vehicle-gateway/cmd/gateway/main_test.go b/go/vehicle-gateway/cmd/gateway/main_test.go index ff1e476e..0c1e5603 100644 --- a/go/vehicle-gateway/cmd/gateway/main_test.go +++ b/go/vehicle-gateway/cmd/gateway/main_test.go @@ -1,6 +1,8 @@ package main import ( + "os" + "strings" "testing" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" @@ -32,6 +34,18 @@ func TestNATSSinkConfigFromEnvUsesExplicitSubjects(t *testing.T) { } } +func TestGatewayConfiguresIdentityLookupCacheTTL(t *testing.T) { + source, err := os.ReadFile("main.go") + if err != nil { + t.Fatalf("read main.go: %v", err) + } + for _, want := range []string{"LookupCacheTTL", "IDENTITY_LOOKUP_CACHE_TTL_SECONDS"} { + if !strings.Contains(string(source), want) { + t.Fatalf("gateway should expose identity lookup cache ttl, missing %s", want) + } + } +} + func TestNATSSinkConfigFromEnvDefaultsToGoSubjects(t *testing.T) { t.Setenv("NATS_URL", "nats://172.17.111.56:4222") diff --git a/go/vehicle-gateway/internal/identity/resolver.go b/go/vehicle-gateway/internal/identity/resolver.go index 2e0aad67..b82a097f 100644 --- a/go/vehicle-gateway/internal/identity/resolver.go +++ b/go/vehicle-gateway/internal/identity/resolver.go @@ -26,12 +26,22 @@ type MySQLResolver struct { db *sql.DB table string locationTouchInterval time.Duration + lookupCacheTTL time.Duration + lookupMu sync.Mutex + lookupCache map[string]lookupCacheEntry touchMu sync.Mutex locationTouches map[string]time.Time } type MySQLResolverOptions struct { LocationTouchInterval time.Duration + LookupCacheTTL time.Duration +} + +type lookupCacheEntry struct { + vin string + notFound bool + expiresAt time.Time } func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver { @@ -50,10 +60,16 @@ func NewMySQLResolverWithOptions(db *sql.DB, table string, opts MySQLResolverOpt if interval <= 0 { interval = 10 * time.Minute } + lookupTTL := opts.LookupCacheTTL + if lookupTTL <= 0 { + lookupTTL = 10 * time.Minute + } return &MySQLResolver{ db: db, table: table, locationTouchInterval: interval, + lookupCacheTTL: lookupTTL, + lookupCache: map[string]lookupCacheEntry{}, locationTouches: map[string]time.Time{}, } } @@ -161,9 +177,32 @@ func (r *MySQLResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope) } func (r *MySQLResolver) lookup(ctx context.Context, column string, value string) (string, error) { + cacheKey := column + "\x00" + value + now := time.Now() + r.lookupMu.Lock() + cached, ok := r.lookupCache[cacheKey] + if ok && now.Before(cached.expiresAt) { + r.lookupMu.Unlock() + if cached.notFound { + return "", sql.ErrNoRows + } + return cached.vin, nil + } + r.lookupMu.Unlock() + query := "SELECT vin FROM " + r.table + " WHERE " + column + " = ? AND vin IS NOT NULL AND vin <> ''" var vin string err := r.db.QueryRowContext(ctx, query, value).Scan(&vin) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return "", err + } + r.lookupMu.Lock() + r.lookupCache[cacheKey] = lookupCacheEntry{ + vin: strings.TrimSpace(vin), + notFound: errors.Is(err, sql.ErrNoRows), + expiresAt: now.Add(r.lookupCacheTTL), + } + r.lookupMu.Unlock() return vin, err } diff --git a/go/vehicle-gateway/internal/identity/resolver_test.go b/go/vehicle-gateway/internal/identity/resolver_test.go index 0e0cb636..730c3da7 100644 --- a/go/vehicle-gateway/internal/identity/resolver_test.go +++ b/go/vehicle-gateway/internal/identity/resolver_test.go @@ -5,6 +5,7 @@ import ( "database/sql" "strings" "testing" + "time" "github.com/DATA-DOG/go-sqlmock" @@ -188,9 +189,6 @@ func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) { WillReturnError(sql.ErrNoRows) mock.ExpectExec("INSERT INTO jt808_registration"). WillReturnResult(sqlmock.NewResult(1, 1)) - mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). - WithArgs("13307795425"). - WillReturnError(sql.ErrNoRows) resolver := NewMySQLResolver(db, "vehicle_identity_binding") for i := 0; i < 2; i++ { @@ -210,6 +208,32 @@ func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) { } } +func TestMySQLResolverCachesIdentityMissesForHighFrequencyFrames(t *testing.T) { + db, mock := newMockDB(t) + defer db.Close() + mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). + WithArgs("13307795425"). + WillReturnError(sql.ErrNoRows) + + resolver := NewMySQLResolverWithOptions(db, "vehicle_identity_binding", MySQLResolverOptions{ + LocationTouchInterval: time.Hour, + LookupCacheTTL: time.Hour, + }) + for i := 0; i < 2; i++ { + _, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + Phone: "013307795425", + Parsed: map[string]any{}, + }) + if err != nil { + t.Fatalf("Resolve() error = %v", err) + } + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestMySQLResolverEnsuresMinimalIdentitySchema(t *testing.T) { db, mock := newMockDB(t) defer db.Close()