diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index 30e6fba9..844c599f 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -51,6 +51,8 @@ type TCPServerConfig struct { MaxConnections int } +const frameOperationTimeout = 30 * time.Second + func NewTCPServer(cfg TCPServerConfig) (*TCPServer, error) { if cfg.Protocol.Protocol == "" { return nil, errors.New("protocol is required") @@ -178,6 +180,9 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) { } func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, source string) { + frameCtx, cancelFrame := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout) + defer cancelFrame() + receivedAtMS := time.Now().UnixMilli() env, err := s.protocol.Parse(raw, receivedAtMS, source) if err != nil { @@ -192,7 +197,7 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, } env.EventID = env.StableEventID() } else { - resolved, resolveErr := s.resolver.Resolve(ctx, env) + resolved, resolveErr := s.resolver.Resolve(frameCtx, env) if resolveErr != nil { s.logger.Warn("identity resolve failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", resolveErr) if env.Parsed == nil { @@ -205,14 +210,14 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, } } - if err := s.sink.PublishRaw(ctx, env); err != nil { + if err := s.sink.PublishRaw(frameCtx, env); err != nil { s.logger.Error("publish raw failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err) return } if env.ParseStatus == envelope.ParseBadFrame { return } - if err := s.sink.PublishUnified(ctx, env); err != nil { + if err := s.sink.PublishUnified(frameCtx, env); err != nil { s.logger.Error("publish unified failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err) return } diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index f9b1e647..cda0b6b3 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -105,6 +105,85 @@ func TestTCPServerWritesProtocolResponseAfterPublish(t *testing.T) { <-done } +func TestTCPServerUsesUncancelledFrameContextForAlreadyReadFrame(t *testing.T) { + parent, cancel := context.WithCancel(context.Background()) + cancel() + + resolver := &contextCheckingResolver{} + sink := &contextCheckingSink{} + server, err := NewTCPServer(TCPServerConfig{ + Protocol: TCPProtocol{ + Protocol: envelope.ProtocolJT808, + Addr: ":0", + Extract: jt808.ExtractFrames, + Parse: func(_ []byte, receivedAtMS int64, sourceEndpoint string) (envelope.FrameEnvelope, error) { + return envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0200", + Phone: "13307795425", + SourceEndpoint: sourceEndpoint, + ReceivedAtMS: receivedAtMS, + EventTimeMS: receivedAtMS, + ParseStatus: envelope.ParseOK, + }, nil + }, + }, + Sink: sink, + Resolver: resolver, + Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), + }) + if err != nil { + t.Fatalf("NewTCPServer() error = %v", err) + } + + server.handleFrame(parent, nil, []byte{0x01}, "127.0.0.1:808") + + if resolver.ctxErr != nil { + t.Fatalf("resolver saw cancelled context: %v", resolver.ctxErr) + } + if sink.rawCtxErr != nil { + t.Fatalf("raw publish saw cancelled context: %v", sink.rawCtxErr) + } + if sink.unifiedCtxErr != nil { + t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr) + } + if sink.rawCount != 1 || sink.unifiedCount != 1 { + t.Fatalf("raw=%d unified=%d", sink.rawCount, sink.unifiedCount) + } +} + +type contextCheckingResolver struct { + ctxErr error +} + +func (r *contextCheckingResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) { + r.ctxErr = ctx.Err() + return env, r.ctxErr +} + +type contextCheckingSink struct { + rawCtxErr error + unifiedCtxErr error + rawCount int + unifiedCount int +} + +func (s *contextCheckingSink) PublishRaw(ctx context.Context, _ envelope.FrameEnvelope) error { + s.rawCtxErr = ctx.Err() + s.rawCount++ + return s.rawCtxErr +} + +func (s *contextCheckingSink) PublishUnified(ctx context.Context, _ envelope.FrameEnvelope) error { + s.unifiedCtxErr = ctx.Err() + s.unifiedCount++ + return s.unifiedCtxErr +} + +func (s *contextCheckingSink) Close() error { + return nil +} + func newTestServer(t *testing.T, protocol TCPProtocol, sink *recordingSink) *TCPServer { t.Helper() server, err := NewTCPServer(TCPServerConfig{