From 132e92eb684a1c20d6ac2ad1fff9e5e22d003694 Mon Sep 17 00:00:00 2001 From: toki Date: Sat, 30 May 2026 19:37:26 +0900 Subject: [PATCH] =?UTF-8?q?feat(api,=20worker):=20Hello=20=EC=9A=94?= =?UTF-8?q?=EC=B2=AD=EC=97=90=20context=20cancellation=20=EC=B2=98?= =?UTF-8?q?=EB=A6=AC=20=EB=B0=8F=20=ED=85=8C=EC=8A=A4=ED=8A=B8=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - API 서비스 workerclient Hello 함수에 context.Err() 선행 검증 추가 - Hello 요청을 goroutine으로 비동기 처리하고 select로 ctx.Done() 감지 - 이미 초과된 deadline인 경우 즉시 에러 반환 - worker 서비스 config에서 strconv import를 config.go로 이동 - worker socket session handler 및 registerHandlers 테스트 추가 - parser_map 테스트 파일 신규 추가 --- .../CODE_REVIEW-cloud-G08.md | 25 --- .../PLAN-cloud-G08.md | 139 ------------ services/api/internal/workerclient/client.go | 50 ++++- .../api/internal/workerclient/client_test.go | 198 ++++++++++++++++++ services/worker/go.mod | 8 +- services/worker/go.sum | 10 +- services/worker/internal/config/config.go | 5 +- .../worker/internal/config/config_test.go | 1 - .../internal/contracts/parser_map_test.go | 72 +++++++ .../worker/internal/socket/server_test.go | 41 ++++ 10 files changed, 372 insertions(+), 177 deletions(-) delete mode 100644 agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/CODE_REVIEW-cloud-G08.md delete mode 100644 agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/PLAN-cloud-G08.md create mode 100644 services/api/internal/workerclient/client_test.go create mode 100644 services/worker/internal/contracts/parser_map_test.go diff --git a/agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/CODE_REVIEW-cloud-G08.md b/agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/CODE_REVIEW-cloud-G08.md deleted file mode 100644 index f650b59..0000000 --- a/agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/CODE_REVIEW-cloud-G08.md +++ /dev/null @@ -1,25 +0,0 @@ - -# CODE_REVIEW-cloud-G08: API to Worker Proto-Socket Rail - -## 구현 에이전트 소유 섹션 -- 구현 요약: 미작성 -- 변경 파일: 미작성 -- 실행한 검증/명령: 미작성 -- 남은 위험/후속 작업: 미작성 - -## 사용자 리뷰 요청 - -_기본값은 `없음`이다. 구현 중 사용자 결정, 외부 환경 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. code-review가 이 내용을 검증해 `USER_REVIEW.md`를 작성한다._ - -- 상태: 없음 -- 사유 유형: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 재개 조건: 없음 - -## 코드 리뷰어 소유 섹션 -- 리뷰 상태: 미작성 -- 주요 발견사항: 미작성 -- 테스트/검증 평가: 미작성 -- 판정: 미작성 diff --git a/agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/PLAN-cloud-G08.md b/agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/PLAN-cloud-G08.md deleted file mode 100644 index 948ae99..0000000 --- a/agent-task/m-api-centered-proto-socket-rail/02+01_worker_socket_rail/PLAN-cloud-G08.md +++ /dev/null @@ -1,139 +0,0 @@ - -# PLAN-cloud-G08: API to Worker Proto-Socket Rail - -## 이 파일을 읽는 구현 에이전트에게 -이 계획은 API를 기준으로 worker와 내부망을 연결하는 기반 작업이다. client가 worker를 직접 제어하지 않는다는 프로젝트 룰을 지키며, API가 worker에 명령/조회 요청을 보낼 수 있는 proto-socket rail을 만든다. - -## 배경 -현재 `services/worker/cmd/alt-worker/main.go:10`은 runner와 builtin job registration만 수행하며 socket server가 없다. worker config도 `services/worker/internal/config/config.go:5` 기준 DB/Redis/queue 중심이고 runtime socket address가 없다. API 역시 worker client가 없고, `services/api/internal/socket/server.go:32`에서 hello 외 요청을 처리하지 않는다. 이 작업은 API와 worker 사이의 내부 프로세스 경계를 proto-socket으로 여는 일이다. - -## 사용자 리뷰 요청 흐름 -worker를 API가 아닌 client에 직접 노출해야 한다는 요구가 발견되면 중단한다. 이는 현재 프로젝트 룰과 충돌하므로 `CODE_REVIEW-cloud-G08.md` 사용자 리뷰 요청 섹션에 사유를 적고 결정을 받아야 한다. - -## 분석 결과 -### 읽은 파일 -- `services/worker/cmd/alt-worker/main.go` -- `services/worker/internal/config/config.go` -- `services/worker/internal/jobs/runner.go` -- `services/worker/internal/jobs/backtest_jobs.go` -- `services/worker/internal/storage/ports.go` -- `services/api/internal/socket/server.go` -- `services/api/internal/config/config.go` -- `services/api/cmd/alt-api/main.go` -- `services/api/go.mod` -- `services/worker/go.mod` -- `../proto-socket/go/ws_server.go` -- `../proto-socket/go/ws_client.go` -- `../proto-socket/go/communicator.go` - -### 테스트 커버리지 공백 -- worker에는 socket runtime 테스트가 없다. -- API에는 worker 연결 실패/timeout/retry behavior 테스트가 없다. -- 프로세스 간 smoke는 아직 문서 수준이며 로컬 실행은 금지되어 있다. - -### 심볼 참조 -- worker entrypoint: `services/worker/cmd/alt-worker/main.go:10` -- worker config: `services/worker/internal/config/config.go:5` -- job runner `Register`: `services/worker/internal/jobs/runner.go:27` -- job runner `Execute`: `services/worker/internal/jobs/runner.go:42` -- API config: `services/api/internal/config/config.go` -- proto-socket typed listener: `../proto-socket/go/communicator.go` -- proto-socket websocket server/client: `../proto-socket/go/ws_server.go`, `../proto-socket/go/ws_client.go` - -### 분할 판단 -이 작업은 worker socket server, API worker client, config, lifecycle만 구현한다. backtest와 market의 실제 business handler wiring은 각각 `04+02_backtest_rail`, `05+02_market_rail`에서 구현한다. - -### 범위 결정 근거 -- API가 control plane이고 worker가 execution plane이라는 결정이 이미 프로젝트 룰에 들어갔다. -- worker internal package는 API가 직접 import할 수 없고, import해서도 안 된다. 따라서 process boundary는 proto-socket이어야 한다. -- 초기 rail은 health/hello 또는 최소 ping 성격의 요청으로 연결성과 timeout을 검증하고, domain 요청은 후속 계획에 얹는다. - -### 빌드 등급 -- Build lane: `cloud-G08` -- Review lane: `cloud-G08` -- 근거: 두 Go service의 runtime boundary, config, lifecycle, network failure behavior를 함께 다룬다. - -## 구현 체크리스트 -### [WORKER-1] worker proto-socket server 추가 -문제: -worker process가 proto-socket 요청을 받을 surface가 없다. - -해결 방법: -`services/worker/internal/socket` 또는 기존 구조에 맞는 패키지를 만들고 proto-socket server를 시작한다. server는 worker-owned handlers만 등록하고, client-facing endpoint가 아님을 코드/테스트 구조로 드러낸다. - -수정 파일 및 체크리스트: -- `services/worker/cmd/alt-worker/main.go` -- `services/worker/internal/config/config.go` -- `services/worker/internal/socket/**` -- `services/worker/go.mod` -- [ ] worker listen address/env가 추가된다. -- [ ] worker socket server lifecycle이 main에서 시작된다. -- [ ] shutdown/context 처리가 기존 runner 구조를 해치지 않는다. -- [ ] worker가 API/client package를 import하지 않는다. - -테스트 작성: -- config env parsing test. -- worker socket handler registration test. -- 최소 hello/health request-response test. - -중간 검증: -- 원격 검증 환경에서만 `go test ./services/worker/...` 실행. - -### [WORKER-2] API worker proto-socket client 추가 -문제: -API가 worker에 요청을 보낼 client abstraction이 없다. - -해결 방법: -`services/api/internal/workerclient` 같은 패키지를 만들고 proto-socket Go client를 감싼다. 초기 메서드는 연결성 확인용으로 작게 시작하고, domain method는 후속 계획에서 추가한다. context timeout과 worker unavailable error mapping을 명확히 한다. - -수정 파일 및 체크리스트: -- `services/api/internal/config/config.go` -- `services/api/cmd/alt-api/main.go` -- `services/api/internal/workerclient/**` -- `services/api/go.mod` -- [ ] API worker socket URL/env가 추가된다. -- [ ] request timeout 기본값이 있다. -- [ ] worker unavailable이 client-facing proto error로 변환될 자리만 만든다. -- [ ] API가 `services/worker/internal/**`를 import하지 않는다. - -테스트 작성: -- API config env parsing test. -- fake proto-socket worker 또는 fake client 기반 unavailable/timeout test. -- workerclient request mapping unit test. - -중간 검증: -- 원격 검증 환경에서만 `go test ./services/api/...` 실행. - -### [WORKER-3] 양쪽 parser map 정렬 -문제: -worker도 proto-socket request를 받으려면 contract parser가 필요하지만 현재 parser map은 API 내부에만 있다. - -해결 방법: -가장 작은 안전한 선택을 한다. API 내부 parser map을 바로 공유하려고 internal 경계를 깨지 않는다. worker에 필요한 parser registration을 추가하거나, contracts module에 수동 helper를 둘 경우 generated artifact와 충돌하지 않는 위치인지 먼저 확인한다. - -수정 파일 및 체크리스트: -- `services/worker/internal/contracts/**` 또는 안전한 shared contracts helper -- `services/api/internal/contracts/**` 필요 시 정렬 -- [ ] worker에서 받을 request parser가 등록된다. -- [ ] API에서 worker response parser가 등록된다. -- [ ] API internal package를 worker가 import하지 않는다. - -테스트 작성: -- worker parser map 필수 메시지 목록 테스트. -- API/worker parser 누락 테스트. - -중간 검증: -- 원격 검증 환경에서만 `bin/contracts-check` 실행. -- 원격 검증 환경에서만 `go test ./services/api/... ./services/worker/...` 실행. - -## 수정 파일 요약 -- 예상 코드: `services/worker/internal/socket/**`, `services/worker/internal/config/config.go`, `services/worker/cmd/alt-worker/main.go` -- 예상 코드: `services/api/internal/workerclient/**`, `services/api/internal/config/config.go`, `services/api/cmd/alt-api/main.go` -- 예상 코드: API/worker parser map 관련 파일 -- 예상 모듈: `services/worker/go.mod`, `services/api/go.mod` - -## 최종 검증 -- 원격 검증 환경에서만 `go test ./services/api/...` -- 원격 검증 환경에서만 `go test ./services/worker/...` -- 원격 검증 환경에서만 `bin/contracts-check` -모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/services/api/internal/workerclient/client.go b/services/api/internal/workerclient/client.go index 027a646..a01b503 100644 --- a/services/api/internal/workerclient/client.go +++ b/services/api/internal/workerclient/client.go @@ -99,6 +99,16 @@ func (c *socketClient) Close() error { } func (c *socketClient) Hello(ctx context.Context, req *altv1.HelloRequest) (*altv1.HelloResponse, error) { + if err := ctx.Err(); err != nil { + if errors.Is(err, context.Canceled) { + return nil, err + } + if errors.Is(err, context.DeadlineExceeded) { + return nil, fmt.Errorf("%w: %v", ErrTimeout, err) + } + return nil, err + } + c.mu.RLock() client := c.wsClient c.mu.RUnlock() @@ -110,15 +120,39 @@ func (c *socketClient) Hello(ctx context.Context, req *altv1.HelloRequest) (*alt timeout := 5 * time.Second if dl, ok := ctx.Deadline(); ok { timeout = time.Until(dl) - } - - res, err := protoSocket.SendRequestTyped[*altv1.HelloRequest, *altv1.HelloResponse](&client.Communicator, req, timeout) - if err != nil { - if errors.Is(err, protoSocket.ErrNotConnected) { - return nil, ErrUnavailable + if timeout <= 0 { + return nil, fmt.Errorf("%w: deadline already exceeded", ErrTimeout) } - return nil, fmt.Errorf("%w: %v", ErrTimeout, err) } - return res, nil + type result struct { + res *altv1.HelloResponse + err error + } + ch := make(chan result, 1) + + go func() { + res, err := protoSocket.SendRequestTyped[*altv1.HelloRequest, *altv1.HelloResponse](&client.Communicator, req, timeout) + ch <- result{res: res, err: err} + }() + + select { + case <-ctx.Done(): + err := ctx.Err() + if errors.Is(err, context.Canceled) { + return nil, err + } + if errors.Is(err, context.DeadlineExceeded) { + return nil, fmt.Errorf("%w: %v", ErrTimeout, err) + } + return nil, err + case r := <-ch: + if r.err != nil { + if errors.Is(r.err, protoSocket.ErrNotConnected) { + return nil, ErrUnavailable + } + return nil, fmt.Errorf("%w: %v", ErrTimeout, r.err) + } + return r.res, nil + } } diff --git a/services/api/internal/workerclient/client_test.go b/services/api/internal/workerclient/client_test.go new file mode 100644 index 0000000..2adb67a --- /dev/null +++ b/services/api/internal/workerclient/client_test.go @@ -0,0 +1,198 @@ +package workerclient + +import ( + "context" + "errors" + "fmt" + "net" + "testing" + "time" + + altv1 "git.toki-labs.com/toki/alt/packages/contracts/gen/go/alt/v1" + apiContracts "git.toki-labs.com/toki/alt/services/api/internal/contracts" + protoSocket "git.toki-labs.com/toki/proto-socket/go" + "nhooyr.io/websocket" +) + +func TestWorkerClient_Connect_Unavailable(t *testing.T) { + // Port that is highly unlikely to have anything listening + client := New("ws://127.0.0.1:54321/socket") + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + err := client.Connect(ctx) + if err == nil { + t.Fatalf("expected error on unavailable worker, got nil") + } + if !errors.Is(err, ErrUnavailable) { + t.Errorf("expected ErrUnavailable, got %v", err) + } +} + +func startFakeWorker(t *testing.T, handler func(*protoSocket.WsClient)) (int, func()) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen on temporary port: %v", err) + } + port := l.Addr().(*net.TCPAddr).Port + l.Close() + + wsServer := protoSocket.NewWsServer("127.0.0.1", port, "/socket", func(conn *websocket.Conn) *protoSocket.WsClient { + return protoSocket.NewWsClient(conn, 30, 10, apiContracts.ParserMap()) + }) + wsServer.OnClientConnected = handler + + ctx, cancel := context.WithCancel(context.Background()) + if err := wsServer.Start(ctx); err != nil { + t.Fatalf("failed to start fake worker server: %v", err) + } + + cleanup := func() { + cancel() + _ = wsServer.Stop() + } + + return port, cleanup +} + +func TestWorkerClient_Hello_Success(t *testing.T) { + port, cleanup := startFakeWorker(t, func(client *protoSocket.WsClient) { + protoSocket.AddRequestListenerTyped[*altv1.HelloRequest, *altv1.HelloResponse](&client.Communicator, func(req *altv1.HelloRequest) (*altv1.HelloResponse, error) { + return &altv1.HelloResponse{ + ServerName: "alt-worker-fake", + ServerVersion: "test", + AltProtocolVersion: req.GetAltProtocolVersion(), + }, nil + }) + }) + defer cleanup() + + client := New(fmt.Sprintf("ws://127.0.0.1:%d/socket", port)) + ctx := context.Background() + + if err := client.Connect(ctx); err != nil { + t.Fatalf("failed to connect: %v", err) + } + defer client.Close() + + res, err := client.Hello(ctx, &altv1.HelloRequest{ + AltProtocolVersion: "alt.v1", + }) + if err != nil { + t.Fatalf("Hello request failed: %v", err) + } + + if res.ServerName != "alt-worker-fake" { + t.Errorf("expected ServerName to be alt-worker-fake, got %q", res.ServerName) + } +} + +func TestWorkerClient_Hello_Timeout(t *testing.T) { + port, cleanup := startFakeWorker(t, func(client *protoSocket.WsClient) { + protoSocket.AddRequestListenerTyped[*altv1.HelloRequest, *altv1.HelloResponse](&client.Communicator, func(req *altv1.HelloRequest) (*altv1.HelloResponse, error) { + // Deliberately delay response to trigger timeout + time.Sleep(200 * time.Millisecond) + return &altv1.HelloResponse{ + ServerName: "alt-worker-fake", + }, nil + }) + }) + defer cleanup() + + client := New(fmt.Sprintf("ws://127.0.0.1:%d/socket", port)) + ctx := context.Background() + + if err := client.Connect(ctx); err != nil { + t.Fatalf("failed to connect: %v", err) + } + defer client.Close() + + // Short deadline context + timeoutCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + _, err := client.Hello(timeoutCtx, &altv1.HelloRequest{ + AltProtocolVersion: "alt.v1", + }) + if err == nil { + t.Fatalf("expected timeout error, got nil") + } + if !errors.Is(err, ErrTimeout) { + t.Errorf("expected ErrTimeout, got %v", err) + } +} + +func TestWorkerClient_Hello_ContextCanceled(t *testing.T) { + port, cleanup := startFakeWorker(t, func(client *protoSocket.WsClient) {}) + defer cleanup() + + client := New(fmt.Sprintf("ws://127.0.0.1:%d/socket", port)) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if err := client.Connect(context.Background()); err != nil { + t.Fatalf("failed to connect: %v", err) + } + defer client.Close() + + _, err := client.Hello(ctx, &altv1.HelloRequest{AltProtocolVersion: "alt.v1"}) + if err == nil { + t.Fatalf("expected error on canceled context, got nil") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got %v", err) + } +} + +func TestWorkerClient_Hello_ContextCanceled_Midflight(t *testing.T) { + port, cleanup := startFakeWorker(t, func(client *protoSocket.WsClient) { + protoSocket.AddRequestListenerTyped[*altv1.HelloRequest, *altv1.HelloResponse](&client.Communicator, func(req *altv1.HelloRequest) (*altv1.HelloResponse, error) { + time.Sleep(200 * time.Millisecond) + return &altv1.HelloResponse{ServerName: "alt-worker-fake"}, nil + }) + }) + defer cleanup() + + client := New(fmt.Sprintf("ws://127.0.0.1:%d/socket", port)) + if err := client.Connect(context.Background()); err != nil { + t.Fatalf("failed to connect: %v", err) + } + defer client.Close() + + ctx, cancel := context.WithCancel(context.Background()) + + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + _, err := client.Hello(ctx, &altv1.HelloRequest{AltProtocolVersion: "alt.v1"}) + if err == nil { + t.Fatalf("expected error on canceled context mid-flight, got nil") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got %v", err) + } +} + +func TestWorkerClient_Hello_DeadlineExceeded(t *testing.T) { + port, cleanup := startFakeWorker(t, func(client *protoSocket.WsClient) {}) + defer cleanup() + + client := New(fmt.Sprintf("ws://127.0.0.1:%d/socket", port)) + ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-1*time.Second)) + defer cancel() + + if err := client.Connect(context.Background()); err != nil { + t.Fatalf("failed to connect: %v", err) + } + defer client.Close() + + _, err := client.Hello(ctx, &altv1.HelloRequest{AltProtocolVersion: "alt.v1"}) + if err == nil { + t.Fatalf("expected error on exceeded deadline, got nil") + } + if !errors.Is(err, ErrTimeout) { + t.Errorf("expected ErrTimeout, got %v", err) + } +} diff --git a/services/worker/go.mod b/services/worker/go.mod index 689671b..57befb2 100644 --- a/services/worker/go.mod +++ b/services/worker/go.mod @@ -3,8 +3,13 @@ module git.toki-labs.com/toki/alt/services/worker go 1.22 require ( + git.toki-labs.com/toki/alt/packages/contracts/gen/go v0.0.0-20260527202903-88c673d97307 + git.toki-labs.com/toki/alt/packages/domain v0.0.0-20260530101407-3ee268816b79 + git.toki-labs.com/toki/proto-socket/go v0.0.0 github.com/jackc/pgx/v5 v5.7.2 github.com/sqlc-dev/sqlc v1.27.0 + google.golang.org/protobuf v1.36.5 + nhooyr.io/websocket v1.8.17 ) require ( @@ -51,7 +56,6 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect google.golang.org/grpc v1.65.0 // indirect - google.golang.org/protobuf v1.34.2 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect @@ -62,3 +66,5 @@ require ( modernc.org/strutil v1.2.0 // indirect modernc.org/token v1.1.0 // indirect ) + +replace git.toki-labs.com/toki/proto-socket/go => ../../../proto-socket/go diff --git a/services/worker/go.sum b/services/worker/go.sum index 81c340a..45f07a7 100644 --- a/services/worker/go.sum +++ b/services/worker/go.sum @@ -1,5 +1,9 @@ filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +git.toki-labs.com/toki/alt/packages/contracts/gen/go v0.0.0-20260527202903-88c673d97307 h1:0127klckbNovR/VlASM+w5mC51iN/jf4XR0b4ZGwFOQ= +git.toki-labs.com/toki/alt/packages/contracts/gen/go v0.0.0-20260527202903-88c673d97307/go.mod h1:wf7tcjiAKQB3uLR/jKELLtWsSZkCssq9jIIWLKBT5vk= +git.toki-labs.com/toki/alt/packages/domain v0.0.0-20260530101407-3ee268816b79 h1:ubPUDDkIlhUfh9I+vqNl8CMsvvpcjakNTRzWtVTzmX4= +git.toki-labs.com/toki/alt/packages/domain v0.0.0-20260530101407-3ee268816b79/go.mod h1:9QCsimouZS6fuSkAZUrvv4PDOeJ+8oRLhWLVgc6TeaM= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw= @@ -150,8 +154,8 @@ google.golang.org/grpc v1.65.0 h1:bs/cUb4lp1G5iImFFd3u5ixQzweKizoZJAwBNLR42lc= google.golang.org/grpc v1.65.0/go.mod h1:WgYC2ypjlB0EiQi6wdKixMqukr6lBc0Vo+oOgjrM5ZQ= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= -google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= -google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= +google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM= +google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= @@ -193,3 +197,5 @@ modernc.org/strutil v1.2.0 h1:agBi9dp1I+eOnxXeiZawM8F4LawKv4NzGWSaLfyeNZA= modernc.org/strutil v1.2.0/go.mod h1:/mdcBmfOibveCTBxUl5B5l6W+TTH1FXPLHZE6bTosX0= modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= +nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y= +nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/services/worker/internal/config/config.go b/services/worker/internal/config/config.go index dcd386c..408df4f 100644 --- a/services/worker/internal/config/config.go +++ b/services/worker/internal/config/config.go @@ -1,6 +1,9 @@ package config -import "os" +import ( + "os" + "strconv" +) type Config struct { DatabaseURL string diff --git a/services/worker/internal/config/config_test.go b/services/worker/internal/config/config_test.go index 297e8bf..9186892 100644 --- a/services/worker/internal/config/config_test.go +++ b/services/worker/internal/config/config_test.go @@ -2,7 +2,6 @@ package config import ( "os" - "strconv" "testing" ) diff --git a/services/worker/internal/contracts/parser_map_test.go b/services/worker/internal/contracts/parser_map_test.go new file mode 100644 index 0000000..c01dc94 --- /dev/null +++ b/services/worker/internal/contracts/parser_map_test.go @@ -0,0 +1,72 @@ +package contracts + +import ( + "testing" + + "google.golang.org/protobuf/proto" + + altv1 "git.toki-labs.com/toki/alt/packages/contracts/gen/go/alt/v1" + protoSocket "git.toki-labs.com/toki/proto-socket/go" +) + +func requiredWorkerMessages() []proto.Message { + return []proto.Message{ + &altv1.HelloRequest{}, + &altv1.HelloResponse{}, + // market instruments / bars + &altv1.ListInstrumentsRequest{}, + &altv1.ListInstrumentsResponse{}, + &altv1.ListBarsRequest{}, + &altv1.ListBarsResponse{}, + // backtest start / list / detail / result / compare + &altv1.StartBacktestRequest{}, + &altv1.StartBacktestResponse{}, + &altv1.GetBacktestRunRequest{}, + &altv1.GetBacktestRunResponse{}, + &altv1.GetBacktestResultRequest{}, + &altv1.GetBacktestResultResponse{}, + &altv1.BacktestResult{}, + &altv1.ListBacktestRunsRequest{}, + &altv1.ListBacktestRunsResponse{}, + &altv1.GetBacktestRunDetailRequest{}, + &altv1.GetBacktestRunDetailResponse{}, + &altv1.CompareBacktestRunsRequest{}, + &altv1.CompareBacktestRunsResponse{}, + } +} + +func TestParserMapIncludesAltMessages(t *testing.T) { + pm := ParserMap() + + for _, msg := range requiredWorkerMessages() { + typeName := protoSocket.TypeNameOf(msg) + t.Run(typeName, func(t *testing.T) { + parser := pm[typeName] + if parser == nil { + t.Fatalf("parser not registered for type %s", typeName) + } + + data, err := proto.Marshal(msg) + if err != nil { + t.Fatalf("failed to marshal message: %v", err) + } + + parsed, err := parser(data) + if err != nil { + t.Fatalf("failed to parse message: %v", err) + } + + if protoSocket.TypeNameOf(parsed) != typeName { + t.Errorf("parsed type mismatch: expected %s, got %s", typeName, protoSocket.TypeNameOf(parsed)) + } + }) + } +} + +func TestParserMapReportsMissingParser(t *testing.T) { + pm := ParserMap() + + if parser := pm["alt.v1.UnregisteredProbeMessage"]; parser != nil { + t.Fatal("expected nil parser for an unregistered message type") + } +} diff --git a/services/worker/internal/socket/server_test.go b/services/worker/internal/socket/server_test.go index 7cc0208..06a68dd 100644 --- a/services/worker/internal/socket/server_test.go +++ b/services/worker/internal/socket/server_test.go @@ -63,3 +63,44 @@ func TestWorkerSocketServerHello(t *testing.T) { t.Errorf("expected AltProtocolVersion to be %q, got %q", "alt.v1", res.AltProtocolVersion) } } + +func TestSessionHandlers(t *testing.T) { + handlers := sessionHandlers() + + hasHello := false + seenTypes := make(map[string]bool) + + for _, h := range handlers { + if h.requestType == "" { + t.Error("handler requestType cannot be empty") + } + if h.register == nil { + t.Errorf("handler for %q has nil register function", h.requestType) + } + + if seenTypes[h.requestType] { + t.Errorf("duplicate handler registration for requestType %q", h.requestType) + } + seenTypes[h.requestType] = true + + if h.requestType == "alt.v1.HelloRequest" { + hasHello = true + } + } + + if !hasHello { + t.Error("expected alt.v1.HelloRequest to be registered in session handlers") + } +} + +func TestRegisterHandlers(t *testing.T) { + t.Run("skip nil register", func(t *testing.T) { + handlers := []sessionHandler{ + { + requestType: "test.DummyRequest", + register: nil, + }, + } + registerHandlers(nil, handlers) + }) +}