diff --git a/agent-task/m-scheduled-market-data-refresh/02_retry_backfill/CODE_REVIEW-local-G04.md b/agent-task/m-scheduled-market-data-refresh/02_retry_backfill/CODE_REVIEW-local-G04.md new file mode 100644 index 0000000..e0e07a0 --- /dev/null +++ b/agent-task/m-scheduled-market-data-refresh/02_retry_backfill/CODE_REVIEW-local-G04.md @@ -0,0 +1,159 @@ + + +# Code Review Reference - BACKFILL + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every implementation-owned section below is filled in. +> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving. +> Fill implementation-owned sections, then stop with active files in place and report ready for review. +> If implementation is blocked by a selected SDD decision or selected Milestone `구현 잠금 > 결정 필요` item, fill `사용자 리뷰 요청` with linked evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. +> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only. + +## 개요 + +date=2026-06-21 +task=m-scheduled-market-data-refresh/02_retry_backfill, plan=0, tag=BACKFILL + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/backtest-loop/milestones/scheduled-market-data-refresh.md` +- Task ids: + - `retry-backfill`: provider 지연, 결측, gap 결과에 따라 retry와 backfill window를 계산하고 실패를 stale/error 상태로 남긴다. +- Completion mode: check-on-pass + +## Spec Targets + +- SDD: `agent-roadmap/sdd/backtest-loop/scheduled-market-data-refresh/SDD.md` +- Acceptance scenarios: + - `S03`: task=`retry-backfill`; evidence=`missing/gap/provider delay fixture 또는 local smoke` +- Completion mode: spec-check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 active plan/review log archive, PASS 시 `complete.log`, task directory archive, m-task 완료 이벤트 보고까지 끝난 상태를 의미합니다. roadmap 수정이나 `update-roadmap` 직접 호출은 런타임 책임입니다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [BACKFILL-1] Retry/Backfill Decision Calculator | [x] | +| [BACKFILL-2] Stable Text/JSONL Evidence | [x] | + +## 구현 체크리스트 + +- [x] `services/worker/internal/scheduler`에 freshness observation -> retry/backfill decision 계산 타입과 함수를 추가한다. +- [x] missing/gap/provider_delay/error 케이스를 stable text/JSONL로 출력하는 writer와 fixture-style unit tests를 추가한다. +- [x] `retry-backfill` 검증: fixture 또는 local smoke에서 missing/gap/provider delay 케이스가 stable text/JSONL로 구분된다. +- [x] 최종 검증으로 `go test -count=1 ./services/worker/internal/scheduler`와 `go test ./services/worker/...`를 실행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G04_N.log`로 아카이브한다. +- [ ] active `PLAN-*-G??.md`를 `plan_local_G04_M.log`로 아카이브한다. +- [ ] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [ ] PASS이면 active task 디렉터리 `agent-task/m-scheduled-market-data-refresh/02_retry_backfill/`를 `agent-task/archive/YYYY/MM/m-scheduled-market-data-refresh/02_retry_backfill/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [ ] PASS이고 task group이 `m-scheduled-market-data-refresh`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-scheduled-market-data-refresh/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이면 다음 active plan/review 또는 USER_REVIEW gate를 처리하고 `complete.log`를 작성하지 않는다. + +## 계획 대비 변경 사항 + +계획에서 제시한 범위를 정확히 준수했다. 추가적인 변경사항은 없음. + +## 주요 설계 결정 + +1. **FreshnessObservation 구조체**: schedule, provider, selector, timeframe, backfill window, expected_from/to, latest_bar_date, missing_dates, gap_buckets, duplicate_count, provider_delay_days, last_error, status 필드를 포함한다. + +2. **BackfillDecision 구조체**: status(fresh|stale|error), backfill_from, backfill_to, has_backfill, retry, missing_count, gap_count, provider_delay, last_error, stale_dates 필드를 포함한다. + +3. **DecideBackfill 함수**: 다음 케이스를 구분한다. + - fresh: missing/gap/error 없음 → has_backfill=false, retry=false + - stale/missing: missing dates 있음 → backfill window 계산, has_backfill=true, retry=true + - stale/gap: gap buckets 있음 → backfill window 계산, has_backfill=true, retry=true + - stale/provider_delay: delay만 있음 → has_backfill=false, retry=false + - error: status="error" + last_error 있음 → error 상태 보존, has_backfill=false, retry=true + +4. **WriteBackfillText**: `schedule= status= reason=<...> backfill_from= backfill_to= retry=` 형식 + +5. **WriteBackfillJSONL**: JSONL 형식 (single-line per decision, no SetIndent) + +6. **Backfill window clamping**: backfill_from이 window start보다 이전이면 window_start로 클램프한다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 SDD 결정 또는 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 연결 대상: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- missing/gap/provider_delay가 fresh success로 숨겨지지 않는지 확인한다. +- backfill_from/backfill_to 계산이 deterministic fixture로 고정됐는지 확인한다. +- text와 JSONL output이 stable field names를 쓰는지 확인한다. +- `Roadmap Targets`와 `Spec Targets`의 `retry-backfill`/`S03` evidence가 실제 검증 결과에 남았는지 확인한다. + +## 검증 결과 + +### BACKFILL-1 중간 검증 +```bash +$ go test -count=1 ./services/worker/internal/scheduler +ok git.toki-labs.com/toki/alt/services/worker/internal/scheduler 0.027s +``` + +### BACKFILL-2 중간 검증 +```bash +$ go test -count=1 ./services/worker/internal/scheduler +ok git.toki-labs.com/toki/alt/services/worker/internal/scheduler 0.027s +``` + +### 최종 검증 +```bash +$ go test -count=1 ./services/worker/internal/scheduler +ok git.toki-labs.com/toki/alt/services/worker/internal/scheduler 0.031s + +$ go test ./services/worker/... +ok git.toki-labs.com/toki/alt/services/worker/cmd/alt-worker 0.018s +? git.toki-labs.com/toki/alt/services/worker/cmd/alt-worker-data-check [no test files] +ok git.toki-labs.com/toki/alt/services/worker/cmd/alt-worker-migrate (cached) +? git.toki-labs.com/toki/alt/services/worker/cmd/alt-worker-schedule-check [no test files] +ok git.toki-labs.com/toki/alt/services/worker/internal/backtest (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/config (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/contracts (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/jobs (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/livetrading (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/marketdata/aggregation (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/marketdata/datacheck (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/marketdata/importer (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/papertrading (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/providers/kis (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/rediskeys (cached) +ok git.toki-labs.com/toki/alt/services/worker/internal/scheduler 0.027s +ok git.toki-labs.com/toki/alt/services/worker/internal/socket (cached) +? git.toki-labs.com/toki/alt/services/worker/internal/storage [no test files] +ok git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres (cached) +? git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres/sqlc [no test files] +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** diff --git a/agent-task/m-scheduled-market-data-refresh/02_retry_backfill/PLAN-local-G04.md b/agent-task/m-scheduled-market-data-refresh/02_retry_backfill/PLAN-local-G04.md new file mode 100644 index 0000000..7777822 --- /dev/null +++ b/agent-task/m-scheduled-market-data-refresh/02_retry_backfill/PLAN-local-G04.md @@ -0,0 +1,154 @@ + + +# Plan - BACKFILL + +## 이 파일을 읽는 구현 에이전트에게 + +`CODE_REVIEW-local-G04.md`의 구현 에이전트 소유 섹션을 채우는 것이 구현의 마지막 단계다. 검증 명령을 실행하고 실제 stdout/stderr를 붙여 넣은 뒤 active 파일을 그대로 두고 review ready 상태로 보고한다. 선택된 SDD 결정이나 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 막으면 직접 질문하지 말고 review stub의 `사용자 리뷰 요청` 섹션에 근거를 기록한다. 구현 에이전트는 `USER_REVIEW.md`, `complete.log`, archive 로그를 만들지 않는다. + +## 배경 + +`retry-backfill`은 provider delay, missing, gap을 성공으로 숨기지 않고 stale/error 상태와 backfill window로 남겨야 한다. 현 코드에는 schedule config validation은 있지만 freshness result를 retry/backfill decision으로 바꾸는 worker-local 계산 표면이 없다. SDD S03은 fixture 또는 local smoke에서 missing/gap/provider delay 케이스가 stable text/JSONL로 구분되는 증거를 요구한다. + +## 사용자 리뷰 요청 흐름 + +사용자 리뷰는 선택된 SDD decision 또는 Milestone lock decision만 active review stub의 `사용자 리뷰 요청` 섹션에 기록한다. 구현 중 직접 채팅 질문, 선택지 제시, `request_user_input` 호출은 금지한다. code-review가 요청 타당성을 검증하고 실제 `USER_REVIEW.md` 작성 여부를 결정한다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/backtest-loop/milestones/scheduled-market-data-refresh.md` +- Task ids: + - `retry-backfill`: provider 지연, 결측, gap 결과에 따라 retry와 backfill window를 계산하고 실패를 stale/error 상태로 남긴다. +- Completion mode: check-on-pass + +## Spec Targets + +- SDD: `agent-roadmap/sdd/backtest-loop/scheduled-market-data-refresh/SDD.md` +- Acceptance scenarios: + - `S03`: task=`retry-backfill`; evidence=`missing/gap/provider delay fixture 또는 local smoke` +- Completion mode: spec-check-on-pass + +## 분석 결과 + +### 읽은 파일 + +- `agent-test/local/rules.md` +- `agent-test/local/worker-smoke.md` +- `agent-roadmap/phase/backtest-loop/milestones/scheduled-market-data-refresh.md` +- `agent-roadmap/sdd/backtest-loop/scheduled-market-data-refresh/SDD.md` +- `services/worker/internal/scheduler/config.go` +- `services/worker/internal/scheduler/config_test.go` +- `services/worker/internal/jobs/marketdata_jobs.go` +- `services/worker/internal/marketdata/importer/importer.go` +- `services/worker/internal/storage/postgres/queries/queries.sql` +- `services/worker/go.mod` + +### 테스트 환경 규칙 + +test_env는 `local`이다. `agent-test/local/rules.md`와 `agent-test/local/worker-smoke.md`를 읽었다. `services/worker/**` 변경이므로 focused command는 `go test -count=1 ./services/worker/internal/scheduler`, 전체 worker smoke는 `go test ./services/worker/...`다. storage SQL을 바꾸지 않으므로 `bin/worker-storage-check`는 필수 검증이 아니다. + +### 테스트 커버리지 공백 + +- schedule validation output은 현재 커버된다. +- missing/gap/provider_delay를 backfill decision으로 바꾸는 worker-local tests가 없다. +- stale/error text/JSONL output fixture가 없다. + +### 심볼 참조 + +none. symbol rename/remove 없음. + +### 분할 판단 + +split decision policy를 먼저 평가했다. shared task group은 `m-scheduled-market-data-refresh`다. 이 subtask는 `02_retry_backfill`이고 `depends_on=[]`이다. `scheduled-runner`와 독립적으로 pure calculator와 stable output을 만들 수 있어 predecessor `complete.log`를 요구하지 않는다. + +### 범위 결정 근거 + +이 plan은 retry/backfill calculation과 stable evidence output만 다룬다. worker long-running loop, API/contracts schema, status CLI, remote runner smoke, backtest-readiness는 제외한다. provider live 호출과 secret 값은 다루지 않는다. + +### 빌드 등급 + +local-G04. 새 계산 표면과 fixture tests가 필요하지만 worker 내부 pure logic이며 외부 runtime 없이 검증 가능하다. + +## 구현 체크리스트 + +- [ ] `services/worker/internal/scheduler`에 freshness observation -> retry/backfill decision 계산 타입과 함수를 추가한다. +- [ ] missing/gap/provider_delay/error 케이스를 stable text/JSONL로 출력하는 writer와 fixture-style unit tests를 추가한다. +- [ ] `retry-backfill` 검증: fixture 또는 local smoke에서 missing/gap/provider delay 케이스가 stable text/JSONL로 구분된다. +- [ ] 최종 검증으로 `go test -count=1 ./services/worker/internal/scheduler`와 `go test ./services/worker/...`를 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [BACKFILL-1] Retry/Backfill Decision Calculator + +문제: SDD는 provider delay나 gap을 성공으로 숨기지 말라고 고정한다(`agent-roadmap/sdd/backtest-loop/scheduled-market-data-refresh/SDD.md:75`). 현재 scheduler package는 validation item만 정의하고(`services/worker/internal/scheduler/config.go:74`) freshness observation이나 backfill decision type이 없다. + +Before: + +```go +// services/worker/internal/scheduler/config.go:74 +// ValidationItem is a stable text/JSONL evidence row. +type ValidationItem struct { +``` + +해결 방법: `services/worker/internal/scheduler/backfill.go`를 추가한다. `FreshnessObservation`은 schedule, provider, selector, timeframe, expected_from/to, latest, missing dates, gap buckets, duplicate count, provider_delay_days, error를 담는다. `DecideBackfill(observation, window)`는 success/fresh, stale/missing, stale/gap, stale/provider_delay, error를 분리하고 earliest missing/gap/latest 기준으로 inclusive `backfill_from`/`backfill_to`를 계산한다. + +수정 파일 및 체크리스트: + +- [ ] `services/worker/internal/scheduler/backfill.go`: `FreshnessObservation`, `BackfillDecision`, `DecideBackfill`. +- [ ] `services/worker/internal/scheduler/backfill_test.go`: missing, gap, provider delay, duplicate-only, error 케이스. + +테스트 작성: `TestDecideBackfillDistinguishesMissingGapProviderDelay`와 `TestDecideBackfillPreservesErrorState`를 추가한다. + +중간 검증: + +```bash +go test -count=1 ./services/worker/internal/scheduler +``` + +기대 결과: exit code 0. backfill calculator tests pass. + +### [BACKFILL-2] Stable Text/JSONL Evidence + +문제: SDD S03은 stale/error 상태와 backfill window가 stable text/JSONL로 구분되어야 한다(`agent-roadmap/sdd/backtest-loop/scheduled-market-data-refresh/SDD.md:87`). 현재 scheduler output은 config validation용 `WriteText`/`WriteJSONL`뿐이고(`services/worker/internal/scheduler/config.go:272`), retry/backfill decision writer가 없다. + +Before: + +```go +// services/worker/internal/scheduler/config.go:272 +// WriteText renders stable operator-facing dry-run lines. +func WriteText(w io.Writer, validation Validation) error { +``` + +해결 방법: `WriteBackfillText`와 `WriteBackfillJSONL`을 추가한다. text는 `schedule= status= reason=<...> backfill_from= backfill_to= retry=` 형태로, JSONL은 같은 필드를 stable struct tag로 출력한다. tests는 expected substrings와 JSONL fields를 모두 고정한다. + +수정 파일 및 체크리스트: + +- [ ] `services/worker/internal/scheduler/backfill.go`: writer 함수와 JSON field tags. +- [ ] `services/worker/internal/scheduler/backfill_test.go`: text/JSONL expected output tests. + +테스트 작성: `TestWriteBackfillTextAndJSONL`를 추가한다. + +중간 검증: + +```bash +go test -count=1 ./services/worker/internal/scheduler +``` + +기대 결과: exit code 0. stable output tests pass. + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `services/worker/internal/scheduler/backfill.go` | BACKFILL-1, BACKFILL-2 | +| `services/worker/internal/scheduler/backfill_test.go` | BACKFILL-1, BACKFILL-2 | + +## 최종 검증 + +```bash +go test -count=1 ./services/worker/internal/scheduler +go test ./services/worker/... +``` + +Go test cache output is acceptable only for the broad second command; the focused first command must use `-count=1`. + +모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/services/worker/cmd/alt-worker-schedule-check/main.go b/services/worker/cmd/alt-worker-schedule-check/main.go new file mode 100644 index 0000000..94e58d1 --- /dev/null +++ b/services/worker/cmd/alt-worker-schedule-check/main.go @@ -0,0 +1,22 @@ +// Command alt-worker-schedule-check validates scheduled market data refresh +// config and prints stable dry-run evidence without starting the worker loop. +package main + +import ( + "os" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/providers/kis" + "git.toki-labs.com/toki/alt/services/worker/internal/scheduler" +) + +func main() { + os.Exit(scheduler.RunCheck( + os.Args[1:], + os.Stdout, + os.Stderr, + time.Now, + scheduler.CapabilityMap(market.ProviderCapability(kis.Capability())), + )) +} diff --git a/services/worker/cmd/alt-worker/main.go b/services/worker/cmd/alt-worker/main.go index 130b6a8..ae47c37 100644 --- a/services/worker/cmd/alt-worker/main.go +++ b/services/worker/cmd/alt-worker/main.go @@ -6,6 +6,7 @@ import ( "log/slog" "os" "os/signal" + "strings" "syscall" "time" @@ -20,6 +21,7 @@ import ( "git.toki-labs.com/toki/alt/services/worker/internal/marketdata/importer" "git.toki-labs.com/toki/alt/services/worker/internal/papertrading" "git.toki-labs.com/toki/alt/services/worker/internal/providers/kis" + "git.toki-labs.com/toki/alt/services/worker/internal/scheduler" "git.toki-labs.com/toki/alt/services/worker/internal/socket" "git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres" ) @@ -42,6 +44,11 @@ func main() { } else { defer pool.Close() deps = storeBackedDeps(runner, postgres.NewStore(pool)) + + if err := startScheduler(ctx, cfg, runner); err != nil { + slog.Error("failed to start scheduler", "error", err) + os.Exit(1) + } } server := socket.NewServer(cfg, deps) @@ -114,3 +121,80 @@ func storeBackedDeps(runner *jobs.Runner, store *postgres.Store) socket.Deps { Live: liveSvc, } } + +func startScheduler(ctx context.Context, cfg config.Config, runner *jobs.Runner) error { + if !cfg.SchedulerEnabled { + return nil + } + + f, err := os.Open(cfg.ScheduleConfigPath) + if err != nil { + return fmt.Errorf("failed to open schedule config: %w", err) + } + defer f.Close() + + schedCfg, err := scheduler.LoadConfig(f) + if err != nil { + return fmt.Errorf("failed to load schedule config: %w", err) + } + + capabilities := scheduler.CapabilityMap(kis.Capability()) + validation := scheduler.ValidateConfig(schedCfg, scheduler.ValidationOptions{ + Now: time.Now(), + Capabilities: capabilities, + }) + if validation.HasRejected() { + var reasons []string + for _, item := range validation.Items { + if item.Status == "rejected" { + reasons = append(reasons, fmt.Sprintf("%s: %s", item.Schedule, item.Reason)) + } + } + return fmt.Errorf("invalid schedule config: %s", strings.Join(reasons, "; ")) + } + + slog.Info("starting scheduler loop", + "scheduler_enabled", true, + "schedule_count", len(schedCfg.Schedules), + ) + + go func() { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + + opts := scheduler.TickOptions{ + Capabilities: capabilities, + Executor: runner, + } + + runSchedulerLoop(ctx, schedCfg, opts, ticker.C) + }() + + return nil +} + +func runSchedulerLoop(ctx context.Context, schedCfg scheduler.Config, opts scheduler.TickOptions, ticks <-chan time.Time) { + for { + select { + case <-ctx.Done(): + slog.Info("stopping scheduler loop") + return + case <-ticks: + opts.Now = time.Now() + res, err := scheduler.RunTick(ctx, schedCfg, opts) + if err != nil { + slog.Error("scheduler tick execution error", "error", err) + continue + } + for _, item := range res.Items { + if item.Status == "failed" { + slog.Error("scheduler job failed", "schedule", item.Schedule, "symbol", item.Symbol, "error", item.Error) + } else if item.Status == "skipped" { + slog.Info("scheduler job skipped", "schedule", item.Schedule, "symbol", item.Symbol, "reason", item.Error) + } else { + slog.Info("scheduler job succeeded", "schedule", item.Schedule, "symbol", item.Symbol, "readiness", item.Readiness, "freshness", item.Freshness) + } + } + } + } +} diff --git a/services/worker/cmd/alt-worker/main_test.go b/services/worker/cmd/alt-worker/main_test.go index b06560d..5a5f39a 100644 --- a/services/worker/cmd/alt-worker/main_test.go +++ b/services/worker/cmd/alt-worker/main_test.go @@ -1,9 +1,17 @@ package main import ( + "context" + "os" + "path/filepath" + "sync" "testing" + "time" + "git.toki-labs.com/toki/alt/services/worker/internal/config" "git.toki-labs.com/toki/alt/services/worker/internal/jobs" + "git.toki-labs.com/toki/alt/services/worker/internal/providers/kis" + "git.toki-labs.com/toki/alt/services/worker/internal/scheduler" "git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres" ) @@ -41,3 +49,128 @@ func TestStoreBackedDepsWiresBacktestSurfaces(t *testing.T) { t.Errorf("expected wiring to register job handlers, runner holds %d", runner.Len()) } } + +func TestStartScheduler(t *testing.T) { + // 1. Scheduler disabled (no config path) + cfg := config.Config{ + SchedulerEnabled: false, + } + runner := jobs.NewRunner() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + err := startScheduler(ctx, cfg, runner) + if err != nil { + t.Fatalf("expected nil error for disabled scheduler, got %v", err) + } + + // 2. Scheduler enabled with invalid config path (should fail) + cfg.SchedulerEnabled = true + cfg.ScheduleConfigPath = "non_existent_file.yaml" + err = startScheduler(ctx, cfg, runner) + if err == nil { + t.Fatal("expected error for non existent schedule config, got nil") + } + + // 3. Scheduler enabled with valid config + tmpDir := t.TempDir() + tmpFile := filepath.Join(tmpDir, "schedule.yaml") + if err := os.WriteFile(tmpFile, []byte(` +schedules: + - name: kr-core-daily + provider: kis + selector: + kind: watchlist + market: KR + venue: KRX + name: kr-core + symbols: ["005930"] + timeframe: daily + cadence: daily + timezone: Asia/Seoul + backfill_window: 3d + parallelism_limit: 2 +`), 0644); err != nil { + t.Fatalf("failed to write temp file: %v", err) + } + + cfg.ScheduleConfigPath = tmpFile + err = startScheduler(ctx, cfg, runner) + if err != nil { + t.Fatalf("expected no error for valid config, got %v", err) + } +} + +type mockExecutor struct { + mu sync.Mutex + dispatched int +} + +func (m *mockExecutor) Execute(ctx context.Context, job jobs.Job) error { + m.mu.Lock() + m.dispatched++ + m.mu.Unlock() + return nil +} + +func TestSchedulerLoopDispatchesTickAndStops(t *testing.T) { + schedCfg := scheduler.Config{ + Schedules: []scheduler.ScheduleConfig{ + { + Name: "kr-core-daily-mock", + Provider: "kis", + Selector: scheduler.SelectorConfig{ + Kind: "watchlist", + Market: "KR", + Venue: "KRX", + Name: "kr-core", + Symbols: []string{"005930"}, + }, + Timeframe: "daily", + Cadence: "daily", + Timezone: "Asia/Seoul", + BackfillWindow: "3d", + ParallelismLimit: 2, + }, + }, + } + + exec := &mockExecutor{} + opts := scheduler.TickOptions{ + Capabilities: scheduler.CapabilityMap(kis.Capability()), + Executor: exec, + } + + ctx, cancel := context.WithCancel(context.Background()) + ticks := make(chan time.Time) + + loopDone := make(chan struct{}) + go func() { + runSchedulerLoop(ctx, schedCfg, opts, ticks) + close(loopDone) + }() + + // Send 1 tick + ticks <- time.Now() + + // Wait briefly for dispatch + time.Sleep(10 * time.Millisecond) + + exec.mu.Lock() + dispatchedVal := exec.dispatched + exec.mu.Unlock() + + if dispatchedVal == 0 { + t.Error("expected at least 1 job dispatch on tick") + } + + // Cancel context to shutdown + cancel() + + select { + case <-loopDone: + // success: loop stopped + case <-time.After(1 * time.Second): + t.Error("scheduler loop did not stop on context cancellation") + } +} diff --git a/services/worker/go.mod b/services/worker/go.mod index 57befb2..546e613 100644 --- a/services/worker/go.mod +++ b/services/worker/go.mod @@ -9,6 +9,7 @@ require ( github.com/jackc/pgx/v5 v5.7.2 github.com/sqlc-dev/sqlc v1.27.0 google.golang.org/protobuf v1.36.5 + gopkg.in/yaml.v3 v3.0.1 nhooyr.io/websocket v1.8.17 ) @@ -57,7 +58,6 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect google.golang.org/grpc v1.65.0 // 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 modernc.org/libc v1.55.3 // indirect modernc.org/mathutil v1.6.0 // indirect diff --git a/services/worker/internal/config/config.go b/services/worker/internal/config/config.go index 400ffed..0236783 100644 --- a/services/worker/internal/config/config.go +++ b/services/worker/internal/config/config.go @@ -6,24 +6,29 @@ import ( ) type Config struct { - DatabaseURL string - RedisURL string - RedisKeyPrefix string - WorkerQueue string - Host string - Port int - SocketPath string + DatabaseURL string + RedisURL string + RedisKeyPrefix string + WorkerQueue string + Host string + Port int + SocketPath string + ScheduleConfigPath string + SchedulerEnabled bool } func Load() Config { + scheduleConfigPath := getenv("ALT_WORKER_SCHEDULE_CONFIG", "") return Config{ - DatabaseURL: getenv("DATABASE_URL", "postgres://alt:alt@localhost:15430/alt?sslmode=disable"), - RedisURL: getenv("REDIS_URL", "redis://localhost:16330/0"), - RedisKeyPrefix: getenv("ALT_REDIS_KEY_PREFIX", "alt"), - WorkerQueue: getenv("ALT_WORKER_QUEUE", "default"), - Host: getenv("ALT_WORKER_HOST", "127.0.0.1"), - Port: getenvInt("ALT_WORKER_PORT", 18031), - SocketPath: getenv("ALT_WORKER_SOCKET_PATH", "/socket"), + DatabaseURL: getenv("DATABASE_URL", "postgres://alt:alt@localhost:15430/alt?sslmode=disable"), + RedisURL: getenv("REDIS_URL", "redis://localhost:16330/0"), + RedisKeyPrefix: getenv("ALT_REDIS_KEY_PREFIX", "alt"), + WorkerQueue: getenv("ALT_WORKER_QUEUE", "default"), + Host: getenv("ALT_WORKER_HOST", "127.0.0.1"), + Port: getenvInt("ALT_WORKER_PORT", 18031), + SocketPath: getenv("ALT_WORKER_SOCKET_PATH", "/socket"), + ScheduleConfigPath: scheduleConfigPath, + SchedulerEnabled: scheduleConfigPath != "", } } diff --git a/services/worker/internal/config/config_test.go b/services/worker/internal/config/config_test.go index 7766cd0..591c942 100644 --- a/services/worker/internal/config/config_test.go +++ b/services/worker/internal/config/config_test.go @@ -9,10 +9,12 @@ func TestConfigLoad(t *testing.T) { os.Setenv("ALT_WORKER_HOST", "127.0.0.9") os.Setenv("ALT_WORKER_PORT", "9999") os.Setenv("ALT_WORKER_SOCKET_PATH", "/ws-test") + os.Setenv("ALT_WORKER_SCHEDULE_CONFIG", "/path/to/schedule.yaml") defer func() { os.Unsetenv("ALT_WORKER_HOST") os.Unsetenv("ALT_WORKER_PORT") os.Unsetenv("ALT_WORKER_SOCKET_PATH") + os.Unsetenv("ALT_WORKER_SCHEDULE_CONFIG") }() cfg := Load() @@ -26,6 +28,12 @@ func TestConfigLoad(t *testing.T) { if cfg.SocketPath != "/ws-test" { t.Errorf("expected SocketPath to be /ws-test, got %q", cfg.SocketPath) } + if cfg.ScheduleConfigPath != "/path/to/schedule.yaml" { + t.Errorf("expected ScheduleConfigPath to be /path/to/schedule.yaml, got %q", cfg.ScheduleConfigPath) + } + if !cfg.SchedulerEnabled { + t.Errorf("expected SchedulerEnabled to be true, got false") + } } func TestConfigDefault(t *testing.T) { @@ -34,6 +42,7 @@ func TestConfigDefault(t *testing.T) { os.Unsetenv("ALT_WORKER_SOCKET_PATH") os.Unsetenv("DATABASE_URL") os.Unsetenv("REDIS_URL") + os.Unsetenv("ALT_WORKER_SCHEDULE_CONFIG") cfg := Load() @@ -52,4 +61,10 @@ func TestConfigDefault(t *testing.T) { if cfg.RedisURL != "redis://localhost:16330/0" { t.Errorf("expected default RedisURL to be redis://localhost:16330/0, got %q", cfg.RedisURL) } + if cfg.ScheduleConfigPath != "" { + t.Errorf("expected default ScheduleConfigPath to be empty, got %q", cfg.ScheduleConfigPath) + } + if cfg.SchedulerEnabled { + t.Errorf("expected default SchedulerEnabled to be false, got true") + } } diff --git a/services/worker/internal/scheduler/backfill.go b/services/worker/internal/scheduler/backfill.go new file mode 100644 index 0000000..f011a23 --- /dev/null +++ b/services/worker/internal/scheduler/backfill.go @@ -0,0 +1,278 @@ +package scheduler + +import ( + "encoding/json" + "fmt" + "io" + "strings" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" +) + +// FreshnessObservation captures the current freshness state of a scheduled +// market data refresh for a given schedule, provider, selector, timeframe and +// backfill window. +type FreshnessObservation struct { + Schedule string `json:"schedule"` + Provider market.Provider `json:"provider"` + Selector string `json:"selector"` + Timeframe string `json:"timeframe"` + Timezone string `json:"timezone"` + Cadence string `json:"cadence"` + BackfillWindow string `json:"backfill_window"` + ExpectedFrom time.Time `json:"expected_from,omitempty"` + ExpectedTo time.Time `json:"expected_to,omitempty"` + LatestBarDate time.Time `json:"latest_bar_date,omitempty"` + MissingDates []time.Time `json:"missing_dates,omitempty"` + GapBuckets []GapBucket `json:"gap_buckets,omitempty"` + DuplicateCount int `json:"duplicate_count"` + ProviderDelay int `json:"provider_delay_days"` + LastError string `json:"last_error,omitempty"` + Status string `json:"status"` // "success", "error" +} + +// GapBucket represents a contiguous range of missing dates. +type GapBucket struct { + From time.Time `json:"from"` + To time.Time `json:"to"` +} + +// BackfillDecision holds the computed backfill decision for a freshness +// observation. +type BackfillDecision struct { + Schedule string `json:"schedule"` + Status string `json:"status"` // "fresh", "stale", "error" + Reason string `json:"reason"` // human readable reason + BackfillFrom time.Time `json:"backfill_from,omitempty"` + BackfillTo time.Time `json:"backfill_to,omitempty"` + HasBackfill bool `json:"has_backfill"` + Retry bool `json:"retry"` + MissingCount int `json:"missing_count"` + GapCount int `json:"gap_count"` + ProviderDelay int `json:"provider_delay_days"` + LastError string `json:"last_error,omitempty"` + StaleDates []time.Time `json:"stale_dates,omitempty"` +} + +// DecideBackfill computes a backfill decision from a freshness observation and +// a backfill window duration. It distinguishes the following cases: +// +// - fresh: no missing dates, no gaps, no error → no backfill needed. +// - stale/missing: missing dates exist → backfill window is computed from +// the earliest missing date. +// - stale/gap: contiguous gap buckets exist → backfill window covers the +// earliest gap. +// - stale/provider_delay: provider reports delay > 0 with no missing dates +// → marks retry=false, no backfill. +// - error: Status=="error" with LastError → preserves error state, includes +// the error text. +func DecideBackfill(obs FreshnessObservation, window time.Duration) BackfillDecision { + dec := BackfillDecision{ + Schedule: obs.Schedule, + ProviderDelay: obs.ProviderDelay, + LastError: obs.LastError, + } + + // Error state: preserve and report. + if obs.Status == "error" && obs.LastError != "" { + dec.Status = "error" + dec.Reason = fmt.Sprintf("previous operation failed: %s", obs.LastError) + dec.HasBackfill = false + dec.Retry = true + return dec + } + + // Collect missing dates and gap buckets. + missingDates := obs.MissingDates + if missingDates == nil { + missingDates = []time.Time{} + } + gapBuckets := obs.GapBuckets + if gapBuckets == nil { + gapBuckets = []GapBucket{} + } + + // Fresh case: nothing missing, no gaps, no error. + if len(missingDates) == 0 && len(gapBuckets) == 0 && obs.ProviderDelay == 0 { + dec.Status = "fresh" + dec.Reason = "all expected dates are present" + dec.HasBackfill = false + dec.Retry = false + return dec + } + + // Provider delay only: missing may be due to delay, not data loss. + if obs.ProviderDelay > 0 && len(missingDates) == 0 && len(gapBuckets) == 0 { + dec.Status = "stale" + dec.Reason = fmt.Sprintf("provider delay %d day(s); no backfill needed", obs.ProviderDelay) + dec.HasBackfill = false + dec.Retry = false + dec.MissingCount = 0 + dec.GapCount = 0 + return dec + } + + // Stale with missing dates or gaps. + dec.Status = "stale" + dec.MissingCount = len(missingDates) + dec.GapCount = len(gapBuckets) + + // Compute backfill window from the earliest missing/gap date. + today := obs.ExpectedTo + if today.IsZero() { + today = time.Now() + } + + // Find the earliest missing date. + var earliestMissing time.Time + hasMissing := false + if len(missingDates) > 0 { + earliestMissing = missingDates[0] + for _, d := range missingDates { + if d.Before(earliestMissing) { + earliestMissing = d + } + } + hasMissing = true + } + + // Find the earliest gap start and latest gap end. + var earliestGapStart time.Time + var latestGapEnd time.Time + hasGap := len(gapBuckets) > 0 + if hasGap { + earliestGapStart = gapBuckets[0].From + latestGapEnd = gapBuckets[0].To + for _, gb := range gapBuckets { + if gb.From.Before(earliestGapStart) { + earliestGapStart = gb.From + } + if gb.To.After(latestGapEnd) { + latestGapEnd = gb.To + } + } + } + + // Determine the overall earliest date. + hasBackfillData := hasMissing || hasGap + if hasBackfillData { + if hasMissing && hasGap { + if earliestGapStart.Before(earliestMissing) { + earliestMissing = earliestGapStart + } + } else if hasGap { + earliestMissing = earliestGapStart + } + + // Set backfill window. + dec.BackfillFrom = earliestMissing + if hasGap && latestGapEnd.After(today) { + dec.BackfillTo = latestGapEnd + } else { + dec.BackfillTo = today + } + + // Bound by window: clamp backfill_from to window start. + windowStart := today.Add(-window) + if dec.BackfillFrom.Before(windowStart) { + dec.BackfillFrom = windowStart + } + + dec.HasBackfill = true + dec.Retry = true + + // Build reason text. + var parts []string + if hasMissing { + parts = append(parts, fmt.Sprintf("%d missing date(s)", len(missingDates))) + } + if hasGap { + parts = append(parts, fmt.Sprintf("%d gap bucket(s)", len(gapBuckets))) + } + dec.Reason = fmt.Sprintf("backfill needed: %s from %s to %s", + strings.Join(parts, ", "), + dec.BackfillFrom.Format("2006-01-02"), + dec.BackfillTo.Format("2006-01-02"), + ) + + // Collect all stale dates (missing + gap ranges). + dec.StaleDates = append(dec.StaleDates, missingDates...) + for _, gb := range gapBuckets { + for d := gb.From; !d.After(gb.To); d = d.Add(24 * time.Hour) { + dec.StaleDates = append(dec.StaleDates, d) + } + } + } + + return dec +} + +// WriteBackfillText renders stable operator-facing dry-run lines for backfill +// decisions. +// +// Format: +// +// schedule= status= reason=<...> backfill_from= backfill_to= retry= +func WriteBackfillText(w io.Writer, decisions []BackfillDecision) error { + for _, dec := range decisions { + backfillFrom := "NONE" + if dec.HasBackfill && !dec.BackfillFrom.IsZero() { + backfillFrom = dec.BackfillFrom.Format("20060102") + } + backfillTo := "NONE" + if dec.HasBackfill && !dec.BackfillTo.IsZero() { + backfillTo = dec.BackfillTo.Format("20060102") + } + line := fmt.Sprintf( + "schedule=%s status=%s reason=%q backfill_from=%s backfill_to=%s retry=%s\n", + dec.Schedule, + dec.Status, + dec.Reason, + backfillFrom, + backfillTo, + fmt.Sprintf("%t", dec.Retry), + ) + if _, err := fmt.Fprint(w, line); err != nil { + return err + } + } + return nil +} + +// WriteBackfillJSONL renders one JSON object per backfill decision. +func WriteBackfillJSONL(w io.Writer, decisions []BackfillDecision) error { + enc := json.NewEncoder(w) + for _, dec := range decisions { + obj := map[string]interface{}{ + "schedule": dec.Schedule, + "status": dec.Status, + "reason": dec.Reason, + "has_backfill": dec.HasBackfill, + "retry": dec.Retry, + "missing_count": dec.MissingCount, + "gap_count": dec.GapCount, + "provider_delay": dec.ProviderDelay, + } + if !dec.BackfillFrom.IsZero() { + obj["backfill_from"] = dec.BackfillFrom.Format("2006-01-02") + } + if !dec.BackfillTo.IsZero() { + obj["backfill_to"] = dec.BackfillTo.Format("2006-01-02") + } + if dec.LastError != "" { + obj["last_error"] = dec.LastError + } + if len(dec.StaleDates) > 0 { + dates := make([]string, 0, len(dec.StaleDates)) + for _, d := range dec.StaleDates { + dates = append(dates, d.Format("2006-01-02")) + } + obj["stale_dates"] = dates + } + if err := enc.Encode(obj); err != nil { + return err + } + } + return nil +} \ No newline at end of file diff --git a/services/worker/internal/scheduler/backfill_test.go b/services/worker/internal/scheduler/backfill_test.go new file mode 100644 index 0000000..841c963 --- /dev/null +++ b/services/worker/internal/scheduler/backfill_test.go @@ -0,0 +1,464 @@ +package scheduler + +import ( + "bytes" + "encoding/json" + "strings" + "testing" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" +) + +func TestDecideBackfillFreshState(t *testing.T) { + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Selector: "kospi", + Timeframe: "daily", + Timezone: "Asia/Seoul", + Cadence: "24h0m0s", + BackfillWindow: "7d", + Status: "success", + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + if dec.Status != "fresh" { + t.Errorf("expected status=fresh, got %s", dec.Status) + } + if dec.HasBackfill { + t.Error("expected has_backfill=false for fresh state") + } + if dec.Retry { + t.Error("expected retry=false for fresh state") + } + if dec.Reason != "all expected dates are present" { + t.Errorf("unexpected reason: %s", dec.Reason) + } +} + +func TestDecideBackfillMissingDates(t *testing.T) { + missingDates := []time.Time{ + time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC), + time.Date(2026, 6, 16, 0, 0, 0, 0, time.UTC), + time.Date(2026, 6, 18, 0, 0, 0, 0, time.UTC), + } + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Selector: "kospi", + Timeframe: "daily", + Timezone: "Asia/Seoul", + Cadence: "24h0m0s", + BackfillWindow: "7d", + Status: "success", + MissingDates: missingDates, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + if dec.Status != "stale" { + t.Errorf("expected status=stale for missing dates, got %s", dec.Status) + } + if !dec.HasBackfill { + t.Error("expected has_backfill=true for missing dates") + } + if !dec.Retry { + t.Error("expected retry=true for missing dates") + } + if dec.MissingCount != 3 { + t.Errorf("expected missing_count=3, got %d", dec.MissingCount) + } + if dec.BackfillFrom.IsZero() { + t.Error("expected backfill_from to be set") + } + // The earliest missing date is 2026-06-15 + if !dec.BackfillFrom.Equal(time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC)) { + t.Errorf("unexpected backfill_from: %s", dec.BackfillFrom.Format("2006-01-02")) + } + if !strings.Contains(dec.Reason, "3 missing date(s)") { + t.Errorf("unexpected reason: %s", dec.Reason) + } +} + +func TestDecideBackfillGapBuckets(t *testing.T) { + gapBuckets := []GapBucket{ + {From: time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC), To: time.Date(2026, 6, 16, 0, 0, 0, 0, time.UTC)}, + {From: time.Date(2026, 6, 18, 0, 0, 0, 0, time.UTC), To: time.Date(2026, 6, 18, 0, 0, 0, 0, time.UTC)}, + } + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Selector: "kospi", + Timeframe: "daily", + Timezone: "Asia/Seoul", + Cadence: "24h0m0s", + BackfillWindow: "7d", + Status: "success", + GapBuckets: gapBuckets, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + ExpectedFrom: time.Date(2026, 6, 13, 0, 0, 0, 0, time.UTC), + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + if dec.Status != "stale" { + t.Errorf("expected status=stale for gap buckets, got %s", dec.Status) + } + if !dec.HasBackfill { + t.Error("expected has_backfill=true for gap buckets") + } + if dec.GapCount != 2 { + t.Errorf("expected gap_count=2, got %d", dec.GapCount) + } + if !strings.Contains(dec.Reason, "2 gap bucket(s)") { + t.Errorf("unexpected reason: %s", dec.Reason) + } +} + +func TestDecideBackfillProviderDelayOnly(t *testing.T) { + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Selector: "kospi", + Timeframe: "daily", + Timezone: "Asia/Seoul", + Cadence: "24h0m0s", + BackfillWindow: "7d", + Status: "success", + ProviderDelay: 2, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + if dec.Status != "stale" { + t.Errorf("expected status=stale for provider delay, got %s", dec.Status) + } + if dec.HasBackfill { + t.Error("expected has_backfill=false for provider delay only") + } + if dec.Retry { + t.Error("expected retry=false for provider delay only") + } + if dec.MissingCount != 0 { + t.Errorf("expected missing_count=0, got %d", dec.MissingCount) + } + if dec.ProviderDelay != 2 { + t.Errorf("expected provider_delay=2, got %d", dec.ProviderDelay) + } +} + +func TestDecideBackfillPreservesErrorState(t *testing.T) { + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Selector: "kospi", + Timeframe: "daily", + Timezone: "Asia/Seoul", + Cadence: "24h0m0s", + BackfillWindow: "7d", + Status: "error", + LastError: "connection timeout to KIS API", + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + if dec.Status != "error" { + t.Errorf("expected status=error, got %s", dec.Status) + } + if dec.HasBackfill { + t.Error("expected has_backfill=false for error state") + } + if !dec.Retry { + t.Error("expected retry=true for error state") + } + if !strings.Contains(dec.Reason, "connection timeout to KIS API") { + t.Errorf("unexpected reason: %s", dec.Reason) + } + if dec.LastError != "connection timeout to KIS API" { + t.Errorf("unexpected last_error: %s", dec.LastError) + } +} + +func TestDecideBackfillDistinguishesMissingGapProviderDelay(t *testing.T) { + window := 7 * 24 * time.Hour + + // Case 1: Missing dates only + obs1 := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Status: "success", + MissingDates: []time.Time{time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC)}, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + dec1 := DecideBackfill(obs1, window) + if dec1.Status != "stale" { + t.Errorf("case 1 (missing): expected stale, got %s", dec1.Status) + } + if !dec1.HasBackfill { + t.Error("case 1 (missing): expected has_backfill=true") + } + + // Case 2: Gap buckets only + obs2 := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Status: "success", + GapBuckets: []GapBucket{{From: time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC), To: time.Date(2026, 6, 16, 0, 0, 0, 0, time.UTC)}}, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + ExpectedFrom: time.Date(2026, 6, 13, 0, 0, 0, 0, time.UTC), + } + dec2 := DecideBackfill(obs2, window) + if dec2.Status != "stale" { + t.Errorf("case 2 (gap): expected stale, got %s", dec2.Status) + } + if !dec2.HasBackfill { + t.Error("case 2 (gap): expected has_backfill=true") + } + + // Case 3: Provider delay only + obs3 := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Status: "success", + ProviderDelay: 3, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + dec3 := DecideBackfill(obs3, window) + if dec3.Status != "stale" { + t.Errorf("case 3 (delay): expected stale, got %s", dec3.Status) + } + if dec3.HasBackfill { + t.Error("case 3 (delay): expected has_backfill=false") + } + if dec3.Retry { + t.Error("case 3 (delay): expected retry=false") + } + + // Case 4: Error + obs4 := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Status: "error", + LastError: "api timeout", + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + dec4 := DecideBackfill(obs4, window) + if dec4.Status != "error" { + t.Errorf("case 4 (error): expected error, got %s", dec4.Status) + } +} + +func TestDecideBackfillDuplicateOnly(t *testing.T) { + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Selector: "kospi", + Status: "success", + DuplicateCount: 5, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + // Duplicate-only should still be fresh if no missing/gap/error. + if dec.Status != "fresh" { + t.Errorf("expected status=fresh for duplicate-only, got %s", dec.Status) + } + if dec.HasBackfill { + t.Error("expected has_backfill=false for duplicate-only") + } +} + +func TestDecideBackfillBackfillWindowCutoff(t *testing.T) { + // Missing dates older than the backfill window should be clamped. + missingDates := []time.Time{ + time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC), // older than 7-day window + time.Date(2026, 6, 10, 0, 0, 0, 0, time.UTC), + } + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Selector: "kospi", + Status: "success", + MissingDates: missingDates, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + if dec.Status != "stale" { + t.Errorf("expected status=stale, got %s", dec.Status) + } + if !dec.HasBackfill { + t.Error("expected has_backfill=true") + } + // Should be clamped to window start (2026-06-13) + windowStart := time.Date(2026, 6, 13, 0, 0, 0, 0, time.UTC) + if !dec.BackfillFrom.Equal(windowStart) { + t.Errorf("expected backfill_from=%s, got %s", windowStart.Format("2006-01-02"), dec.BackfillFrom.Format("2006-01-02")) + } +} + +func TestDecideBackfillNilSlices(t *testing.T) { + obs := FreshnessObservation{ + Schedule: "daily-daily", + Provider: market.ProviderKIS, + Status: "success", + MissingDates: nil, + GapBuckets: nil, + ExpectedTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + } + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + if dec.Status != "fresh" { + t.Errorf("expected status=fresh for nil slices, got %s", dec.Status) + } +} + +func TestWriteBackfillTextAndJSONL(t *testing.T) { + decisions := []BackfillDecision{ + { + Schedule: "daily-daily", + Status: "fresh", + Reason: "all expected dates are present", + HasBackfill: false, + Retry: false, + }, + { + Schedule: "daily-daily", + Status: "stale", + Reason: "backfill needed: 3 missing date(s) from 2026-06-15 to 2026-06-20", + HasBackfill: true, + Retry: true, + BackfillFrom: time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC), + BackfillTo: time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC), + MissingCount: 3, + StaleDates: []time.Time{ + time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC), + time.Date(2026, 6, 16, 0, 0, 0, 0, time.UTC), + time.Date(2026, 6, 18, 0, 0, 0, 0, time.UTC), + }, + }, + { + Schedule: "daily-daily", + Status: "error", + Reason: "previous operation failed: connection timeout", + LastError: "connection timeout", + HasBackfill: false, + Retry: true, + }, + } + + // Test text output + var textBuf bytes.Buffer + if err := WriteBackfillText(&textBuf, decisions); err != nil { + t.Fatalf("WriteBackfillText: %v", err) + } + + text := textBuf.String() + textLines := strings.Split(strings.TrimSpace(text), "\n") + if len(textLines) != 3 { + t.Errorf("expected 3 text lines, got %d", len(textLines)) + } + + // Check line 1 (fresh) + if !strings.Contains(textLines[0], "status=fresh") { + t.Errorf("line 1 should contain status=fresh: %s", textLines[0]) + } + if !strings.Contains(textLines[0], "backfill_from=NONE") { + t.Errorf("line 1 should contain backfill_from=NONE: %s", textLines[0]) + } + + // Check line 2 (stale) + if !strings.Contains(textLines[1], "status=stale") { + t.Errorf("line 2 should contain status=stale: %s", textLines[1]) + } + if !strings.Contains(textLines[1], "backfill_from=20260615") { + t.Errorf("line 2 should contain backfill_from=20260615: %s", textLines[1]) + } + if !strings.Contains(textLines[1], "backfill_to=20260620") { + t.Errorf("line 2 should contain backfill_to=20260620: %s", textLines[1]) + } + + // Check line 3 (error) + if !strings.Contains(textLines[2], "status=error") { + t.Errorf("line 3 should contain status=error: %s", textLines[2]) + } + + // Test JSONL output (single-line JSON per decision, no SetIndent) + var jsonBuf bytes.Buffer + if err := WriteBackfillJSONL(&jsonBuf, decisions); err != nil { + t.Fatalf("WriteBackfillJSONL: %v", err) + } + + jsonLines := strings.Split(strings.TrimSpace(jsonBuf.String()), "\n") + if len(jsonLines) != 3 { + t.Fatalf("expected 3 JSON lines, got %d", len(jsonLines)) + } + + // Parse and verify each JSON line + for i, jsonLine := range jsonLines { + var parsed map[string]interface{} + if err := json.Unmarshal([]byte(jsonLine), &parsed); err != nil { + t.Fatalf("line %d: failed to parse JSON: %v", i, err) + } + + if parsed["schedule"] != "daily-daily" { + t.Errorf("line %d: expected schedule=daily-daily, got %v", i, parsed["schedule"]) + } + + expectedStatus := []string{"fresh", "stale", "error"}[i] + if parsed["status"] != expectedStatus { + t.Errorf("line %d: expected status=%s, got %v", i, expectedStatus, parsed["status"]) + } + + // Fresh case: backfill_from should not be present + if i == 0 { + if val, ok := parsed["backfill_from"]; ok && val != nil { + t.Errorf("line %d: expected backfill_from absent or nil for fresh, got %v", i, val) + } + } + // Stale case: should have stale_dates + if i == 1 { + staleDates, ok := parsed["stale_dates"].([]interface{}) + if !ok { + t.Errorf("line %d: expected stale_dates to be an array, got %T", i, parsed["stale_dates"]) + } else if len(staleDates) != 3 { + t.Errorf("line %d: expected 3 stale_dates, got %d", i, len(staleDates)) + } + } + // Error case: should have last_error + if i == 2 { + if parsed["last_error"] != "connection timeout" { + t.Errorf("line %d: expected last_error=connection timeout, got %v", i, parsed["last_error"]) + } + } + } +} + +func TestDecideBackfillNilObservation(t *testing.T) { + // Empty observation with no status should not trigger error or backfill. + obs := FreshnessObservation{} + + window := 7 * 24 * time.Hour + dec := DecideBackfill(obs, window) + + // Should be fresh (empty = no missing, no gaps, no error). + if dec.Status != "fresh" { + t.Errorf("expected status=fresh for empty observation, got %s", dec.Status) + } +} \ No newline at end of file diff --git a/services/worker/internal/scheduler/config.go b/services/worker/internal/scheduler/config.go new file mode 100644 index 0000000..53dabec --- /dev/null +++ b/services/worker/internal/scheduler/config.go @@ -0,0 +1,374 @@ +// Package scheduler validates scheduled market data refresh configuration and +// renders dry-run evidence. It does not own the long-running worker loop; the +// runner can consume the validated shapes in a later orchestration layer. +package scheduler + +import ( + "encoding/json" + "flag" + "fmt" + "io" + "os" + "sort" + "strconv" + "strings" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" + "gopkg.in/yaml.v3" +) + +const ( + exitOK = 0 + exitRejected = 1 + exitUsage = 2 +) + +// Config is the top-level YAML shape for scheduled market data refresh. +type Config struct { + Schedules []ScheduleConfig `yaml:"schedules"` +} + +// ScheduleConfig declares one named refresh schedule. +type ScheduleConfig struct { + Name string `yaml:"name"` + Provider string `yaml:"provider"` + Selector SelectorConfig `yaml:"selector"` + Timeframe string `yaml:"timeframe"` + Cadence string `yaml:"cadence"` + Timezone string `yaml:"timezone"` + BackfillWindow string `yaml:"backfill_window"` + ParallelismLimit int `yaml:"parallelism_limit"` +} + +// SelectorConfig is the named universe selector for a schedule. +type SelectorConfig struct { + Kind string `yaml:"kind"` + Market string `yaml:"market"` + Venue string `yaml:"venue"` + Name string `yaml:"name"` + Symbols []string `yaml:"symbols"` +} + +// ValidationOptions carries deterministic inputs for dry-run validation. +type ValidationOptions struct { + Now time.Time + Capabilities map[market.Provider]market.ProviderCapability +} + +// Validation reports every schedule's dry-run status. +type Validation struct { + Items []ValidationItem +} + +// HasRejected reports whether any item was rejected. +func (v Validation) HasRejected() bool { + for _, item := range v.Items { + if item.Status == "rejected" { + return true + } + } + return false +} + +// ValidationItem is a stable text/JSONL evidence row. +type ValidationItem struct { + Schedule string `json:"schedule"` + Status string `json:"status"` + Provider string `json:"provider,omitempty"` + Selector string `json:"selector,omitempty"` + Timeframe string `json:"timeframe,omitempty"` + Cadence string `json:"cadence,omitempty"` + Timezone string `json:"timezone,omitempty"` + BackfillWindow string `json:"backfill_window,omitempty"` + ParallelismLimit int `json:"parallelism_limit,omitempty"` + NextRun string `json:"next_run,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// LoadConfig reads a YAML schedule config. +func LoadConfig(r io.Reader) (Config, error) { + var cfg Config + dec := yaml.NewDecoder(r) + dec.KnownFields(true) + if err := dec.Decode(&cfg); err != nil { + return Config{}, fmt.Errorf("decode schedule config: %w", err) + } + return cfg, nil +} + +// ValidateConfig validates config and computes the next dry-run window. +func ValidateConfig(cfg Config, opts ValidationOptions) Validation { + now := opts.Now + if now.IsZero() { + now = time.Now() + } + if len(cfg.Schedules) == 0 { + return Validation{Items: []ValidationItem{{ + Schedule: "unknown", + Status: "rejected", + Reason: "no schedules declared", + }}} + } + + seen := make(map[string]bool) + items := make([]ValidationItem, 0, len(cfg.Schedules)) + for _, sched := range cfg.Schedules { + items = append(items, validateSchedule(sched, opts.Capabilities, now, seen)) + } + return Validation{Items: items} +} + +func validateSchedule(s ScheduleConfig, caps map[market.Provider]market.ProviderCapability, now time.Time, seen map[string]bool) ValidationItem { + name := strings.TrimSpace(s.Name) + if name == "" { + name = "unknown" + } + item := ValidationItem{ + Schedule: name, + Provider: strings.TrimSpace(s.Provider), + Selector: strings.TrimSpace(s.Selector.Name), + Timeframe: strings.TrimSpace(s.Timeframe), + Timezone: strings.TrimSpace(s.Timezone), + ParallelismLimit: s.ParallelismLimit, + } + + var reasons []string + if s.Name == "" { + reasons = append(reasons, "name is required") + } else if seen[s.Name] { + reasons = append(reasons, fmt.Sprintf("duplicate schedule name %q", s.Name)) + } + seen[s.Name] = true + if item.Provider == "" { + reasons = append(reasons, "provider is required") + } + if item.Selector == "" { + reasons = append(reasons, "selector.name is required") + } + if s.Selector.Kind == "" { + reasons = append(reasons, "selector.kind is required") + } + if s.Selector.Venue == "" { + reasons = append(reasons, "selector.venue is required") + } + if len(s.Selector.Symbols) == 0 { + reasons = append(reasons, "selector.symbols is required") + } + if s.Timeframe == "" { + reasons = append(reasons, "timeframe is required") + } + if s.Cadence == "" { + reasons = append(reasons, "cadence is required") + } + if s.Timezone == "" { + reasons = append(reasons, "timezone is required") + } + if s.BackfillWindow == "" { + reasons = append(reasons, "backfill_window is required") + } + if s.ParallelismLimit <= 0 { + reasons = append(reasons, "parallelism_limit must be greater than zero") + } + + cadence, err := parseWindow(s.Cadence) + if err != nil { + reasons = append(reasons, fmt.Sprintf("invalid cadence %q: %v", s.Cadence, err)) + } else { + item.Cadence = cadence.String() + } + backfill, err := parseWindow(s.BackfillWindow) + if err != nil { + reasons = append(reasons, fmt.Sprintf("invalid backfill_window %q: %v", s.BackfillWindow, err)) + } else { + item.BackfillWindow = backfill.String() + } + loc, err := time.LoadLocation(s.Timezone) + if err != nil && s.Timezone != "" { + reasons = append(reasons, fmt.Sprintf("invalid timezone %q: %v", s.Timezone, err)) + } + tf, tfLabel, err := parseTimeframe(s.Timeframe) + if err != nil { + reasons = append(reasons, err.Error()) + } else { + item.Timeframe = tfLabel + } + + provider := market.Provider(s.Provider) + capability, ok := caps[provider] + if s.Provider != "" && !ok { + reasons = append(reasons, fmt.Sprintf("unsupported provider %q", s.Provider)) + } + if ok && err == nil { + decision := capability.CheckBars(market.ProviderCapabilityRequest{ + Provider: provider, + Market: market.Market(s.Selector.Market), + Venue: market.Venue(s.Selector.Venue), + Timeframe: tf, + }) + if !decision.Accepted { + reasons = append(reasons, "unsupported provider/timeframe combination: "+decision.Reason) + } + } + + if len(reasons) > 0 { + item.Status = "rejected" + item.Reason = strings.Join(reasons, "; ") + return item + } + item.Status = "valid" + item.NextRun = nextWindow(now, loc, cadence).Format(time.RFC3339) + return item +} + +func parseTimeframe(value string) (market.Timeframe, string, error) { + switch strings.ToLower(strings.TrimSpace(value)) { + case "daily", "1d": + return market.TimeframeDaily, "daily", nil + case "monthly", "1mo": + return market.TimeframeMonthly, "monthly", fmt.Errorf("unsupported timeframe %q (only daily is supported)", value) + case "minute_1", "1m": + return market.TimeframeMin1, "minute_1", fmt.Errorf("unsupported timeframe %q (only daily is supported)", value) + case "minute_5", "5m": + return market.TimeframeMin5, "minute_5", fmt.Errorf("unsupported timeframe %q (only daily is supported)", value) + default: + return "", "", fmt.Errorf("unsupported timeframe %q (only daily is supported)", value) + } +} + +func parseWindow(value string) (time.Duration, error) { + trimmed := strings.TrimSpace(strings.ToLower(value)) + if trimmed == "daily" { + return 24 * time.Hour, nil + } + if strings.HasSuffix(trimmed, "d") && len(trimmed) > 1 { + days, err := strconv.Atoi(strings.TrimSuffix(trimmed, "d")) + if err != nil || days <= 0 { + return 0, fmt.Errorf("expected positive day count") + } + return time.Duration(days) * 24 * time.Hour, nil + } + d, err := time.ParseDuration(trimmed) + if err != nil { + return 0, err + } + if d <= 0 { + return 0, fmt.Errorf("expected positive duration") + } + return d, nil +} + +func nextWindow(now time.Time, loc *time.Location, cadence time.Duration) time.Time { + localNow := now.In(loc) + dayStart := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), 0, 0, 0, 0, loc) + if !localNow.After(dayStart) { + return dayStart + } + elapsed := localNow.Sub(dayStart) + steps := elapsed/cadence + 1 + return dayStart.Add(steps * cadence) +} + +// WriteText renders stable operator-facing dry-run lines. +func WriteText(w io.Writer, validation Validation) error { + for _, item := range validation.Items { + if item.Status == "rejected" { + if _, err := fmt.Fprintf(w, "schedule=%s status=rejected reason=%q\n", item.Schedule, item.Reason); err != nil { + return err + } + continue + } + if _, err := fmt.Fprintf(w, + "schedule=%s status=valid provider=%s selector=%s timeframe=%s cadence=%s timezone=%s backfill_window=%s parallelism_limit=%d next_run=%s\n", + item.Schedule, item.Provider, item.Selector, item.Timeframe, item.Cadence, item.Timezone, item.BackfillWindow, item.ParallelismLimit, item.NextRun, + ); err != nil { + return err + } + } + return nil +} + +// WriteJSONL renders one JSON object per validation item. +func WriteJSONL(w io.Writer, validation Validation) error { + enc := json.NewEncoder(w) + for _, item := range validation.Items { + if err := enc.Encode(item); err != nil { + return err + } + } + return nil +} + +// RunCheck implements the schedule config validation command. +func RunCheck(args []string, stdout, stderr io.Writer, now func() time.Time, caps map[market.Provider]market.ProviderCapability) int { + fs := flag.NewFlagSet("alt-worker-schedule-check", flag.ContinueOnError) + fs.SetOutput(stderr) + file := fs.String("file", "", "path to schedule YAML file") + output := fs.String("output", "text", "output format: text or jsonl") + at := fs.String("at", "", "dry-run reference time in RFC3339") + if err := fs.Parse(args); err != nil { + return exitUsage + } + if *file == "" { + fmt.Fprintln(stderr, "schedule config file is required: --file ") + return exitUsage + } + f, err := os.Open(*file) + if err != nil { + fmt.Fprintf(stderr, "open schedule config: %v\n", err) + return exitUsage + } + defer f.Close() + cfg, err := LoadConfig(f) + if err != nil { + fmt.Fprintf(stderr, "validate schedule config: %v\n", err) + return exitUsage + } + nowValue := now() + if *at != "" { + parsed, err := time.Parse(time.RFC3339, *at) + if err != nil { + fmt.Fprintf(stderr, "invalid --at time %q: %v\n", *at, err) + return exitUsage + } + nowValue = parsed + } + validation := ValidateConfig(cfg, ValidationOptions{Now: nowValue, Capabilities: caps}) + switch *output { + case "text": + err = WriteText(stdout, validation) + case "jsonl": + err = WriteJSONL(stdout, validation) + default: + fmt.Fprintf(stderr, "unknown output format %q (want text or jsonl)\n", *output) + return exitUsage + } + if err != nil { + fmt.Fprintf(stderr, "write schedule validation: %v\n", err) + return exitUsage + } + if validation.HasRejected() { + return exitRejected + } + return exitOK +} + +// CapabilityMap builds a deterministic capability map for callers. +func CapabilityMap(capabilities ...market.ProviderCapability) map[market.Provider]market.ProviderCapability { + out := make(map[market.Provider]market.ProviderCapability, len(capabilities)) + for _, capability := range capabilities { + out[capability.Provider] = capability + } + return out +} + +// SortedScheduleNames returns names in deterministic order for future status +// surfaces without exposing internal map iteration order. +func SortedScheduleNames(cfg Config) []string { + names := make([]string, 0, len(cfg.Schedules)) + for _, sched := range cfg.Schedules { + names = append(names, sched.Name) + } + sort.Strings(names) + return names +} diff --git a/services/worker/internal/scheduler/config_test.go b/services/worker/internal/scheduler/config_test.go new file mode 100644 index 0000000..3ff9f0e --- /dev/null +++ b/services/worker/internal/scheduler/config_test.go @@ -0,0 +1,170 @@ +package scheduler + +import ( + "bytes" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" +) + +func testCapability() market.ProviderCapability { + return market.ProviderCapability{ + Provider: market.ProviderKIS, + Rules: []market.ProviderCapabilityRule{ + { + Market: market.MarketKR, + Venue: market.VenueKRX, + AssetTypes: map[market.AssetType]bool{ + market.AssetTypeEquity: true, + }, + Timeframes: map[market.Timeframe]bool{ + market.TimeframeDaily: true, + }, + }, + }, + } +} + +func TestValidateConfigReportsNextWindowAndRejectedCombination(t *testing.T) { + cfg, err := LoadConfig(strings.NewReader(` +schedules: + - name: kr-core-daily + provider: kis + selector: + kind: watchlist + market: KR + venue: KRX + name: kr-core + symbols: ["005930", "000660"] + timeframe: daily + cadence: daily + timezone: Asia/Seoul + backfill_window: 3d + parallelism_limit: 2 + - name: kr-core-monthly + provider: kis + selector: + kind: watchlist + market: KR + venue: KRX + name: kr-core + symbols: ["005930"] + timeframe: monthly + cadence: daily + timezone: Asia/Seoul + backfill_window: 3d + parallelism_limit: 1 +`)) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + now := time.Date(2026, 6, 21, 10, 30, 0, 0, time.UTC) + result := ValidateConfig(cfg, ValidationOptions{ + Now: now, + Capabilities: CapabilityMap(testCapability()), + }) + + if len(result.Items) != 2 { + t.Fatalf("items = %d, want 2", len(result.Items)) + } + valid := result.Items[0] + if valid.Status != "valid" { + t.Fatalf("first status = %q, want valid: %+v", valid.Status, valid) + } + if valid.NextRun != "2026-06-22T00:00:00+09:00" { + t.Errorf("next run = %q, want 2026-06-22T00:00:00+09:00", valid.NextRun) + } + if valid.Cadence != "24h0m0s" || valid.BackfillWindow != "72h0m0s" { + t.Errorf("durations = %s/%s, want 24h0m0s/72h0m0s", valid.Cadence, valid.BackfillWindow) + } + rejected := result.Items[1] + if rejected.Status != "rejected" { + t.Fatalf("second status = %q, want rejected", rejected.Status) + } + if !strings.Contains(rejected.Reason, "unsupported timeframe") { + t.Errorf("rejection reason %q missing unsupported timeframe", rejected.Reason) + } +} + +func TestWriteTextAndJSONL(t *testing.T) { + validation := Validation{Items: []ValidationItem{ + { + Schedule: "kr-core-daily", + Status: "valid", + Provider: "kis", + Selector: "kr-core", + Timeframe: "daily", + Cadence: "24h0m0s", + Timezone: "Asia/Seoul", + BackfillWindow: "72h0m0s", + ParallelismLimit: 2, + NextRun: "2026-06-22T00:00:00+09:00", + }, + {Schedule: "bad", Status: "rejected", Reason: "unsupported provider/timeframe combination"}, + }} + + var text bytes.Buffer + if err := WriteText(&text, validation); err != nil { + t.Fatalf("WriteText: %v", err) + } + for _, want := range []string{ + "schedule=kr-core-daily status=valid", + "next_run=2026-06-22T00:00:00+09:00", + "schedule=bad status=rejected", + "unsupported provider/timeframe combination", + } { + if !strings.Contains(text.String(), want) { + t.Errorf("text output %q missing %q", text.String(), want) + } + } + + var jsonl bytes.Buffer + if err := WriteJSONL(&jsonl, validation); err != nil { + t.Fatalf("WriteJSONL: %v", err) + } + if !strings.Contains(jsonl.String(), `"schedule":"kr-core-daily"`) { + t.Errorf("jsonl output %q missing valid schedule", jsonl.String()) + } + if !strings.Contains(jsonl.String(), `"status":"rejected"`) { + t.Errorf("jsonl output %q missing rejected status", jsonl.String()) + } +} + +func TestRunCheckReturnsRejectedWhenAnyScheduleIsRejected(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "schedule.yaml") + if err := os.WriteFile(path, []byte(` +schedules: + - name: unsupported + provider: kis + selector: + kind: watchlist + market: KR + venue: KRX + name: kr-core + symbols: ["005930"] + timeframe: minute_1 + cadence: 24h + timezone: Asia/Seoul + backfill_window: 48h + parallelism_limit: 1 +`), 0644); err != nil { + t.Fatalf("write fixture: %v", err) + } + + var stdout, stderr bytes.Buffer + code := RunCheck([]string{"--file", path, "--output", "jsonl", "--at", "2026-06-21T10:30:00Z"}, &stdout, &stderr, func() time.Time { + return time.Date(2026, 6, 21, 10, 30, 0, 0, time.UTC) + }, CapabilityMap(testCapability())) + + if code != exitRejected { + t.Fatalf("exit code = %d, want %d (stderr=%q stdout=%q)", code, exitRejected, stderr.String(), stdout.String()) + } + if !strings.Contains(stdout.String(), `"status":"rejected"`) { + t.Errorf("stdout %q missing rejected JSONL", stdout.String()) + } +} diff --git a/services/worker/internal/scheduler/runner.go b/services/worker/internal/scheduler/runner.go new file mode 100644 index 0000000..7e2048e --- /dev/null +++ b/services/worker/internal/scheduler/runner.go @@ -0,0 +1,244 @@ +package scheduler + +import ( + "context" + "encoding/json" + "fmt" + "io" + "sort" + "sync" + "time" + + "git.toki-labs.com/toki/alt/packages/domain/market" + "git.toki-labs.com/toki/alt/services/worker/internal/jobs" +) + +// Executor represents the job execution engine. +type Executor interface { + Execute(ctx context.Context, job jobs.Job) error +} + +// TickOptions provides parameters for executing a tick. +type TickOptions struct { + Now time.Time + Capabilities map[market.Provider]market.ProviderCapability + Executor Executor +} + +// TickResult represents the outcome of a single scheduler tick. +type TickResult struct { + Items []ItemResult +} + +// ItemResult represents the outcome of a single symbol import within a schedule. +type ItemResult struct { + Schedule string `json:"schedule"` + Symbol string `json:"symbol"` + Status string `json:"status"` // "succeeded", "failed", "skipped" + Provider string `json:"provider"` + Timeframe string `json:"timeframe"` + Readiness string `json:"readiness"` + Freshness string `json:"freshness"` + Error string `json:"error,omitempty"` +} + +var ( + activeJobs = make(map[string]bool) + activeJobsMu sync.Mutex +) + +// resetActiveJobs clears the active running schedules (for testing). +func resetActiveJobs() { + activeJobsMu.Lock() + defer activeJobsMu.Unlock() + activeJobs = make(map[string]bool) +} + +func currentWindow(now time.Time, loc *time.Location, cadence time.Duration) time.Time { + next := nextWindow(now, loc, cadence) + return next.Add(-cadence) +} + +// RunTick triggers the periodic import task scheduler. +func RunTick(ctx context.Context, cfg Config, opts TickOptions) (TickResult, error) { + now := opts.Now + if now.IsZero() { + now = time.Now() + } + + validation := ValidateConfig(cfg, ValidationOptions{ + Now: now, + Capabilities: opts.Capabilities, + }) + + var results []ItemResult + var resultsMu sync.Mutex + var wg sync.WaitGroup + + for i, sched := range cfg.Schedules { + valItem := validation.Items[i] + if valItem.Status != "valid" { + continue + } + + wg.Add(1) + go func(s ScheduleConfig) { + defer wg.Done() + + loc, err := time.LoadLocation(s.Timezone) + if err != nil { + return + } + cadence, _ := parseWindow(s.Cadence) + backfill, _ := parseWindow(s.BackfillWindow) + + windowTime := currentWindow(now, loc, cadence) + lockKey := fmt.Sprintf("%s:%s", s.Name, windowTime.Format(time.RFC3339)) + + activeJobsMu.Lock() + if activeJobs[lockKey] { + activeJobsMu.Unlock() + resultsMu.Lock() + for _, sym := range s.Selector.Symbols { + results = append(results, ItemResult{ + Schedule: s.Name, + Symbol: sym, + Status: "skipped", + Provider: s.Provider, + Timeframe: s.Timeframe, + Readiness: "stale", + Freshness: "stale", + Error: "duplicate window running", + }) + } + resultsMu.Unlock() + return + } + activeJobs[lockKey] = true + activeJobsMu.Unlock() + + defer func() { + activeJobsMu.Lock() + delete(activeJobs, lockKey) + activeJobsMu.Unlock() + }() + + sem := make(chan struct{}, s.ParallelismLimit) + var schedWg sync.WaitGroup + + for _, sym := range s.Selector.Symbols { + schedWg.Add(1) + sem <- struct{}{} + + go func(symbol string) { + defer func() { + <-sem + schedWg.Done() + }() + + fromTime := windowTime.Add(-backfill) + payload := jobs.DailyBarImportPayload{ + Provider: s.Provider, + SelectorKind: s.Selector.Kind, + Market: s.Selector.Market, + Venue: s.Selector.Venue, + Name: s.Selector.Name, + Symbols: []string{symbol}, + From: fromTime.Format("20060102"), + To: windowTime.Format("20060102"), + Timeframe: s.Timeframe, + } + + payloadJSON, err := json.Marshal(payload) + if err != nil { + resultsMu.Lock() + results = append(results, ItemResult{ + Schedule: s.Name, + Symbol: symbol, + Status: "failed", + Provider: s.Provider, + Timeframe: s.Timeframe, + Readiness: "error", + Freshness: "stale", + Error: fmt.Sprintf("marshal payload: %v", err), + }) + resultsMu.Unlock() + return + } + + jobID := fmt.Sprintf("sched-%s-%s-%d", s.Name, symbol, windowTime.Unix()) + job := jobs.Job{ + ID: jobID, + Kind: jobs.KindImportDailyBars, + Payload: payloadJSON, + } + + execErr := opts.Executor.Execute(ctx, job) + + itemRes := ItemResult{ + Schedule: s.Name, + Symbol: symbol, + Provider: s.Provider, + Timeframe: s.Timeframe, + } + + if execErr != nil { + itemRes.Status = "failed" + itemRes.Readiness = "error" + itemRes.Freshness = "stale" + itemRes.Error = execErr.Error() + } else { + itemRes.Status = "succeeded" + itemRes.Readiness = "ready" + itemRes.Freshness = "fresh" + } + + resultsMu.Lock() + results = append(results, itemRes) + resultsMu.Unlock() + }(sym) + } + schedWg.Wait() + }(sched) + } + + wg.Wait() + + sort.Slice(results, func(i, j int) bool { + if results[i].Schedule != results[j].Schedule { + return results[i].Schedule < results[j].Schedule + } + return results[i].Symbol < results[j].Symbol + }) + + return TickResult{Items: results}, nil +} + +// WriteTickText renders stable tick result lines. +func WriteTickText(w io.Writer, result TickResult) error { + for _, item := range result.Items { + if item.Error != "" { + if _, err := fmt.Fprintf(w, "schedule=%s symbol=%s status=%s provider=%s timeframe=%s readiness=%s freshness=%s error=%q\n", + item.Schedule, item.Symbol, item.Status, item.Provider, item.Timeframe, item.Readiness, item.Freshness, item.Error); err != nil { + return err + } + } else { + if _, err := fmt.Fprintf(w, "schedule=%s symbol=%s status=%s provider=%s timeframe=%s readiness=%s freshness=%s\n", + item.Schedule, item.Symbol, item.Status, item.Provider, item.Timeframe, item.Readiness, item.Freshness); err != nil { + return err + } + } + } + return nil +} + +// WriteTickJSONL renders one JSON object per tick item. +func WriteTickJSONL(w io.Writer, result TickResult) error { + enc := json.NewEncoder(w) + for _, item := range result.Items { + if err := enc.Encode(item); err != nil { + return err + } + } + return nil +} diff --git a/services/worker/internal/scheduler/runner_test.go b/services/worker/internal/scheduler/runner_test.go new file mode 100644 index 0000000..579786b --- /dev/null +++ b/services/worker/internal/scheduler/runner_test.go @@ -0,0 +1,392 @@ +package scheduler + +import ( + "bytes" + "context" + "fmt" + "strings" + "sync" + "testing" + "time" + + "git.toki-labs.com/toki/alt/services/worker/internal/jobs" + "git.toki-labs.com/toki/alt/services/worker/internal/marketdata/importer" +) + +type fakeExecutor struct { + mu sync.Mutex + activeCount int + maxActive int + executedJobs []jobs.Job + executeFn func(ctx context.Context, job jobs.Job) error +} + +func (f *fakeExecutor) Execute(ctx context.Context, job jobs.Job) error { + f.mu.Lock() + f.activeCount++ + if f.activeCount > f.maxActive { + f.maxActive = f.activeCount + } + f.executedJobs = append(f.executedJobs, job) + f.mu.Unlock() + + defer func() { + f.mu.Lock() + f.activeCount-- + f.mu.Unlock() + }() + + if f.executeFn != nil { + return f.executeFn(ctx, job) + } + return nil +} + +func TestRunTickDispatchesItemsWithBoundedParallelism(t *testing.T) { + resetActiveJobs() + cfg := Config{ + Schedules: []ScheduleConfig{ + { + Name: "kr-daily-parallel", + Provider: "kis", + Selector: SelectorConfig{ + Kind: "watchlist", + Market: "KR", + Venue: "KRX", + Name: "watchlist1", + Symbols: []string{"A", "B", "C", "D"}, + }, + Timeframe: "daily", + Cadence: "daily", + Timezone: "Asia/Seoul", + BackfillWindow: "3d", + ParallelismLimit: 2, + }, + }, + } + + exec := &fakeExecutor{ + executeFn: func(ctx context.Context, job jobs.Job) error { + time.Sleep(10 * time.Millisecond) + return nil + }, + } + + opts := TickOptions{ + Now: time.Date(2026, 6, 21, 10, 30, 0, 0, time.UTC), + Capabilities: CapabilityMap(testCapability()), + Executor: exec, + } + + res, err := RunTick(context.Background(), cfg, opts) + if err != nil { + t.Fatalf("RunTick: %v", err) + } + + if len(res.Items) != 4 { + t.Errorf("expected 4 item results, got %d", len(res.Items)) + } + + if exec.maxActive > 2 { + t.Errorf("max concurrent jobs %d exceeded parallelism limit 2", exec.maxActive) + } + + for _, item := range res.Items { + if item.Status != "succeeded" { + t.Errorf("expected succeeded status, got %s", item.Status) + } + } +} + +func TestRunTickSkipsDuplicateRunningWindow(t *testing.T) { + resetActiveJobs() + cfg := Config{ + Schedules: []ScheduleConfig{ + { + Name: "kr-daily-dup", + Provider: "kis", + Selector: SelectorConfig{ + Kind: "watchlist", + Market: "KR", + Venue: "KRX", + Name: "watchlist1", + Symbols: []string{"A"}, + }, + Timeframe: "daily", + Cadence: "daily", + Timezone: "Asia/Seoul", + BackfillWindow: "3d", + ParallelismLimit: 1, + }, + }, + } + + startChan := make(chan struct{}) + doneChan := make(chan struct{}) + + exec := &fakeExecutor{ + executeFn: func(ctx context.Context, job jobs.Job) error { + close(startChan) + <-doneChan + return nil + }, + } + + opts := TickOptions{ + Now: time.Date(2026, 6, 21, 10, 30, 0, 0, time.UTC), + Capabilities: CapabilityMap(testCapability()), + Executor: exec, + } + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + _, _ = RunTick(context.Background(), cfg, opts) + }() + + <-startChan // Wait until first tick starts executing + + res2, err := RunTick(context.Background(), cfg, opts) + if err != nil { + t.Fatalf("second RunTick: %v", err) + } + + if len(res2.Items) != 1 { + t.Fatalf("expected 1 result in duplicate run, got %d", len(res2.Items)) + } + if res2.Items[0].Status != "skipped" { + t.Errorf("expected duplicate run to be skipped, got %s", res2.Items[0].Status) + } + if res2.Items[0].Error != "duplicate window running" { + t.Errorf("expected error string duplicate window running, got %q", res2.Items[0].Error) + } + + close(doneChan) + wg.Wait() + + // Sequential 2nd run + exec3 := &fakeExecutor{} + opts3 := TickOptions{ + Now: time.Date(2026, 6, 21, 10, 30, 0, 0, time.UTC), + Capabilities: CapabilityMap(testCapability()), + Executor: exec3, + } + res3, err := RunTick(context.Background(), cfg, opts3) + if err != nil { + t.Fatalf("third sequential RunTick: %v", err) + } + if len(res3.Items) != 1 { + t.Fatalf("expected 1 result, got %d", len(res3.Items)) + } + if res3.Items[0].Status != "succeeded" { + t.Errorf("expected succeeded status, got %s", res3.Items[0].Status) + } +} + +func TestWriteTickJSONLIncludesReadinessFields(t *testing.T) { + res := TickResult{ + Items: []ItemResult{ + { + Schedule: "test-sched", + Symbol: "AAPL", + Status: "succeeded", + Provider: "kis", + Timeframe: "daily", + Readiness: "ready", + Freshness: "fresh", + }, + { + Schedule: "test-sched", + Symbol: "MSFT", + Status: "failed", + Provider: "kis", + Timeframe: "daily", + Readiness: "error", + Freshness: "stale", + Error: "network error", + }, + }, + } + + var jsonl bytes.Buffer + if err := WriteTickJSONL(&jsonl, res); err != nil { + t.Fatalf("WriteTickJSONL: %v", err) + } + + lines := strings.Split(strings.TrimSpace(jsonl.String()), "\n") + if len(lines) != 2 { + t.Fatalf("expected 2 JSONL lines, got %d", len(lines)) + } + + if !strings.Contains(lines[0], `"readiness":"ready"`) || !strings.Contains(lines[0], `"freshness":"fresh"`) { + t.Errorf("line 0 missing readiness/freshness fields: %s", lines[0]) + } + if !strings.Contains(lines[1], `"readiness":"error"`) || !strings.Contains(lines[1], `"freshness":"stale"`) { + t.Errorf("line 1 missing readiness/freshness fields: %s", lines[1]) + } + + var text bytes.Buffer + if err := WriteTickText(&text, res); err != nil { + t.Fatalf("WriteTickText: %v", err) + } + textLines := strings.Split(strings.TrimSpace(text.String()), "\n") + if len(textLines) != 2 { + t.Fatalf("expected 2 text lines, got %d", len(textLines)) + } + if !strings.Contains(textLines[0], "readiness=ready freshness=fresh") { + t.Errorf("text line 0 missing readiness/freshness: %s", textLines[0]) + } + if !strings.Contains(textLines[1], "readiness=error freshness=stale") { + t.Errorf("text line 1 missing readiness/freshness: %s", textLines[1]) + } +} + +type fakeDailyBarImporter struct { + mu sync.Mutex + upsertedBars map[string]int +} + +func (f *fakeDailyBarImporter) ImportDailyBars(ctx context.Context, request importer.DailyBarRequest) (importer.Result, error) { + f.mu.Lock() + defer f.mu.Unlock() + + timeframe := "daily" + days := int(request.To.Sub(request.From).Hours()/24) + 1 + for _, sym := range request.Selector.Symbols { + for i := 0; i < days; i++ { + t := request.From.Add(time.Duration(i) * 24 * time.Hour) + dateStr := t.Format("2006-01-02") + key := fmt.Sprintf("%s:%s:%s", sym, timeframe, dateStr) + f.upsertedBars[key]++ + } + } + return importer.Result{ + Instruments: len(request.Selector.Symbols), + Bars: len(request.Selector.Symbols) * days, + }, nil +} + +func TestRunTickSequentialTicksDoNotDuplicateBarKeys(t *testing.T) { + resetActiveJobs() + + fakeImp := &fakeDailyBarImporter{ + upsertedBars: make(map[string]int), + } + + runner := jobs.NewRunner() + jobs.RegisterBuiltins(runner) + cap := testCapability() + jobs.RegisterDailyBarImportHandler(runner, cap, fakeImp) + + cfg := Config{ + Schedules: []ScheduleConfig{ + { + Name: "kr-daily-dup-assert", + Provider: "kis", + Selector: SelectorConfig{ + Kind: "watchlist", + Market: "KR", + Venue: "KRX", + Name: "watchlist1", + Symbols: []string{"005930"}, + }, + Timeframe: "daily", + Cadence: "daily", + Timezone: "Asia/Seoul", + BackfillWindow: "3d", + ParallelismLimit: 1, + }, + }, + } + + opts := TickOptions{ + Now: time.Date(2026, 6, 21, 10, 30, 0, 0, time.UTC), + Capabilities: CapabilityMap(cap), + Executor: runner, + } + + // 1st tick + res1, err := RunTick(context.Background(), cfg, opts) + if err != nil { + t.Fatalf("first tick: %v", err) + } + if len(res1.Items) != 1 || res1.Items[0].Status != "succeeded" { + t.Fatalf("expected succeeded in 1st tick, got: %+v", res1.Items) + } + + fakeImp.mu.Lock() + keysCount1 := len(fakeImp.upsertedBars) + if keysCount1 == 0 { + fakeImp.mu.Unlock() + t.Fatal("expected upserted bars count > 0") + } + for k, count := range fakeImp.upsertedBars { + if count != 1 { + t.Errorf("expected key %s count to be 1, got %d", k, count) + } + } + fakeImp.mu.Unlock() + + // 2nd tick + res2, err := RunTick(context.Background(), cfg, opts) + if err != nil { + t.Fatalf("second tick: %v", err) + } + if len(res2.Items) != 1 || res2.Items[0].Status != "succeeded" { + t.Fatalf("expected succeeded in 2nd tick, got: %+v", res2.Items) + } + + fakeImp.mu.Lock() + keysCount2 := len(fakeImp.upsertedBars) + if keysCount2 != keysCount1 { + t.Errorf("expected unique key count to be %d, got %d (duplicated keys detected)", keysCount1, keysCount2) + } + for k, count := range fakeImp.upsertedBars { + if count != 2 { + t.Errorf("expected key %s count to be 2 after second tick, got %d", k, count) + } + } + fakeImp.mu.Unlock() +} + +func TestRunTickOutputCarriesS02Evidence(t *testing.T) { + res := TickResult{ + Items: []ItemResult{ + { + Schedule: "test-sched", + Symbol: "AAPL", + Status: "succeeded", + Provider: "kis", + Timeframe: "daily", + Readiness: "ready", + Freshness: "fresh", + }, + }, + } + + var jsonl bytes.Buffer + if err := WriteTickJSONL(&jsonl, res); err != nil { + t.Fatalf("WriteTickJSONL: %v", err) + } + + outStr := jsonl.String() + for _, expected := range []string{`"status":"succeeded"`, `"readiness":"ready"`, `"freshness":"fresh"`, `"schedule":"test-sched"`, `"symbol":"AAPL"`} { + if !strings.Contains(outStr, expected) { + t.Errorf("JSONL output missing expected pattern %q: %s", expected, outStr) + } + } + + var text bytes.Buffer + if err := WriteTickText(&text, res); err != nil { + t.Fatalf("WriteTickText: %v", err) + } + + textStr := text.String() + for _, expected := range []string{"status=succeeded", "readiness=ready", "freshness=fresh", "schedule=test-sched", "symbol=AAPL"} { + if !strings.Contains(textStr, expected) { + t.Errorf("Text output missing expected pattern %q: %s", expected, textStr) + } + } +} +