Alphea App

Phase 1 완료 — StreamManager + ConnectNode handshake¶

commit:server|d84dd9f

bidi 핸드쉐이크 절반을 먼저 착륙. 노드 토큰 강제 / 인메모리 세션 레지스트리 / ConnectedMsg 송신 / heartbeat ack round-trip 까지 동작. JobAssign·Complete 흐름은 Phase 2 에서 wiring.

변경¶

• src/domain/node/stream.rs (신규) — StreamManager (DashMap of node_id → mpsc tx).
  • register / unregister / send_to / touch_heartbeat / snapshot / connected_count / is_connected
  • 동일 node_id 재연결 시 이전 세션의 tx 를 drop → 옛 응답 스트림이 깔끔히 종료.
  • NodeStatusKind { Active, Busy, Draining, Inactive } (현재 phase 는 Active 만 사용).
• src/domain/node/mod.rs (신규) — 모듈 루트, pub use re-export.
• src/domain/mod.rs — pub mod node; 추가.
• src/server/mod.rs — AppState.stream_manager: Arc<StreamManager> 추가, run() 에서 초기화.
• src/main.rs — AppState 구성에 stream_manager 주입.
• src/handler/node.rs::connect_node — 본문 구현:
  1. caller(req)?.token_type == "node" 강제 (불일치 시 permission_denied).
  2. nodes::get_by_id 로 tier / max_concurrent_jobs / owner_wallet_address 로드.
  3. StreamManager::register → mpsc Receiver 획득.
  4. ConnectedMsg { node_id, session_id (UUID v4), server_time } 송신.
  5. inbound reader task spawn — NodeHeartbeatMsg 수신 시 touch_heartbeat + DB nodes::touch_heartbeat + HeartbeatAckMsg 송신. StatusUpdate / JobComplete / JobFailed 는 Phase 2 까지 로그만.
  6. 스트림 종료 시 unregister.

인증¶

/alphea.v1.NodeService/ConnectNode 는 allow-list 에 추가하지 않음 — 글로벌 AuthLayer 가 토큰을 검증하고 핸들러 본문에서 node-token 을 강제하는 구조. JWT sub 가 곧 node_id.

검증¶

• cargo check clean.
• cargo clippy --lib Phase 1 신규 코드 경고 0.
• cargo test --lib 62/62 통과 (StreamManager 단위 테스트 5건 포함):
  • register_then_send_delivers_message
  • reregister_drops_previous_session
  • send_to_unknown_node_returns_false
  • unregister_removes_node
  • touch_heartbeat_updates_running_jobs

남은 Phase¶

Phase 2~8 은 본 이슈에 그대로 — Phase 2 (QueuedJob + Dispatcher 코어) 가 다음 작업.

Phase 2 완료 — QueuedJob + Dispatcher core¶

commit:server|6fb9580

bidi 파이프라인의 엔진 절반 착륙. Dispatcher 가 QueuedJob 배치를 받아 적합 노드의 outbound 스트림에 JobAssignMsg 를 송신하고, 노드의 JobComplete / JobFailed 콜백이 다시 dispatcher 로 routing 됨.

변경¶

• src/engine/job.rs (신규) — QueuedJob (JobAssignMsg 가 필요로 하는 모든 필드 보유), JobQueue (Mutex<VecDeque> 기반 drop-newest 정책 + received / dropped 카운터), generate_job_id (legacy 호환 base36 형식).
• src/engine/dispatcher.rs (신규) — Dispatcher, Assignment, DispatchOutcome.
  • dispatch(batch) — 연결 노드 0 이면 전부 fallback 큐로. 그 외에는 StreamManager::snapshot() 순회 → tier 매칭 + 가용 슬롯 확인 → 첫 적합 노드에 JobAssignMsg 송신 → assignments map 등록.
  • handle_job_complete / handle_job_failed — DashMap<job_id, Assignment> 에서 제거 + 로그. Phase 5 에서 receipt + AGX 정산 + BalanceUpdateMsg push 추가 예정.
  • tier_covers — node tier ≥ job required tier (proto enum 숫자 순서 활용).
  • timeout 30s 는 Assignment.deadline 으로 기록만, sweep 은 Phase 6.
• src/engine/mod.rs (신규) — re-export.
• src/lib.rs — pub mod engine;.
• src/server/mod.rs — AppState.dispatcher: Arc<Dispatcher> + fallback JobQueue 초기화.
• src/main.rs — 동일 wiring.
• src/handler/node.rs::handle_node_message — NodeMessage::JobComplete / JobFailed 를 state.dispatcher.handle_* 로 forwarding (Phase 1 의 로그 only 코드 대체).
• src/admin/mod.rs — admin REST 2개 추가:
  • POST /api/engine/jobs { count, type, tier?, priority?, base_reward_agx? } — 합성 job 배치 생성 → Dispatcher::dispatch 호출. count 1..=1000, type ∈ {compute, storage, ai, stream}, tier 기본 low.
  • GET /api/engine/snapshot — connected_nodes, pending_assignments, fallback_queue.{queued,received,dropped} 노출 (Phase 8 에서 synth/scheduler stats 까지 확장 예정).

검증¶

• cargo check clean.
• cargo clippy --lib Phase 2 신규 코드 경고 0.
• cargo test --lib 71/71 통과 (engine 9건 추가):
  • engine::job::enqueue_then_dequeue_returns_in_order
  • engine::job::enqueue_past_capacity_increments_dropped
  • engine::job::dequeue_more_than_present_returns_what_exists
  • engine::job::generate_job_id_unique_and_well_formed
  • engine::dispatcher::dispatch_with_no_nodes_falls_back_to_queue
  • engine::dispatcher::dispatch_picks_eligible_node_and_records_assignment
  • engine::dispatcher::dispatch_skips_nodes_below_required_tier
  • engine::dispatcher::handle_complete_clears_assignment
  • engine::dispatcher::node_message_complete_is_constructible

Phase 2 의도적 단순화 (이후 phase 에서 보강)¶

• 단일 노드 할당 (Phase 6 에서 multi-node redundancy / FindNodesForJob).
• Activeness 점수, INACTIVE 자동 전이 없음 (Phase 6).
• timeout sweep 없음 (Phase 6).
• 영속화 없음 — receipt / aggregated_earnings 갱신은 Phase 5.
• fallback_queue consumer 없음 — Scheduler 가 Phase 4 에서 drain 시작.
• Synth 자동 루프 없음 — admin REST 로만 트리거 (Phase 3).

Phase 4 완료 — Scheduler + EdgeNode bootstrap¶

commit:server|8e349e3

(Phase 3 Synth 보다 먼저 진행 — 이후 Phase 3 가 이 fallback 큐를 자동으로 채우게 됨.)

엔진의 fallback 절반 착륙. dispatcher 가 실 노드 매칭 실패로 큐에 쌓은 job 을 built-in edge 노드 시뮬레이터가 tick 으로 drain.

변경¶

• src/engine/edge_node.rs (신규) — EdgeNodeInfo, derive_edge_wallet (SHA-256 + jwt_secret salt, 끝 20 bytes hex), ensure_edge_node. legacy Go 와 byte-for-byte 동일.
• src/engine/scheduler.rs (신규) — Scheduler (atomic 기반 control flags), tick 루프 (기본 500ms), tier-weighted 실패 시뮬 (HIGH 1% / MID 2% / LOW 5%), SchedulerSnapshot.
  • enabled / running / tick_ms / jobs_per_tick 모두 atomic — admin 의 pause/resume 이 tick 중간에도 race 없이 적용.
  • 영속화는 Phase 5 — 현재는 in-memory 카운터만 (total_assigned, total_completed, total_failed).
• src/store/nodes.rs::ensure_edge — INSERT … ON CONFLICT DO UPDATE 로 edge 노드 row idempotent upsert. node_type=EDGE, auth_mode=WALLET, status=ACTIVE, archived_at=NULL 강제 (재시작 시 즉시 복귀).
• src/server/mod.rs + src/main.rs — 부팅 시 ensure_edge_node + Scheduler::spawn 호출, AppState 에 scheduler / edge 추가.
• src/admin/mod.rs:
  • GET /api/engine/snapshot — scheduler stats (enabled/running/tick_ms/jobs_per_tick/queue stats/total counters) + edge config (node_id/wallet/tier/capacity/score) 포함하도록 확장.
  • POST /api/engine/scheduler/control { action, value? } — start / stop / pause / resume / set_tick / set_batch. 서버측 clamp (tick 1002000ms, batch 10010000).

검증¶

• cargo build clean.
• cargo clippy --lib Phase 4 신규 코드 경고 0.
• cargo test --lib 78/78 통과 (engine 14건, 5건 추가):
  • engine::edge_node::derive_edge_wallet_is_deterministic_and_well_formed
  • engine::edge_node::from_salt_uses_hardcoded_constants
  • engine::scheduler::tick_drains_queue_and_increments_counters
  • engine::scheduler::tick_respects_jobs_per_tick_cap
  • engine::scheduler::pause_disables_tick_processing
  • engine::scheduler::set_tick_ms_clamps_to_range
  • engine::scheduler::set_jobs_per_tick_clamps_to_range

다음 phase 로 미룬 항목¶

• Phase 3 (Synth): 자동 job 생성 루프 — 이 fallback 큐를 채우게 됨.
• Phase 5: scheduler 가 처리한 job 의 receipt + aggregated_earnings 적립.
• Phase 6: timeout sweep + activeness 페널티.

Phase 3 완료 — Synth (합성 job 생성기)¶

commit:server|5ee923c

엔진의 demand 절반 착륙. tick 으로 시나리오 가중치 + burst + 백프레셔를 반영해 batch 를 만들고 dispatcher 에 push. 이제 Phase 1~4 만으로도 데모 흐름 (Synth → Dispatcher → 연결 노드 또는 Scheduler fallback) 이 자동으로 돌아간다.

변경¶

• src/engine/synth.rs (신규) — Synth, Scenario enum, scenario 별 가중치 테이블, BurstState, SynthSnapshot.
  • 기본 100ms tick. update_burst (확률 3% 진입, 20–50 tick × 2–3× multiplier).
  • calculate_job_count — bootstrap rate max(1000, active_nodes×20) 기반. measured process_rate 가 100 초과면 그쪽으로 전환. fallback queue backlog > 10초 분량이면 절반 감속 (burst 중에는 미적용).
  • 시나리오별 가중치 — Balanced(40/25/20/15), ComputeHeavy(60/20/10/10), AiHeavy(20/15/50/15), Stress(default + burst params).
  • per-job 도출 — derive_required_tier (compute=LOW, storage=70% LOW/30% MID, ai=60% MID/40% HIGH, stream=MID), derive_priority (base + burst bonus + jitter, clamp 1..=10), derive_expected_exec_ms (type 별 range × 0.7..1.3 jitter), derive_complexity (AI burst 시 2.0).
  • !Send 처리 — build_batch 를 sync 헬퍼로 분리해 rand::thread_rng() 가 dispatcher.dispatch().await 경계를 넘지 않도록.
• 운영 안전 기본값: enabled=false 로 시작 — 프로덕션이 부팅 즉시 데모 부하를 자동 생성하지 않도록. admin REST resume 으로 명시 활성.
• src/engine/mod.rs — re-export.
• src/server/mod.rs + src/main.rs — AppState.synth + 부팅 시 Synth::spawn.
• src/admin/mod.rs:
  • GET /api/engine/snapshot — synth 섹션 추가 (enabled/tick_ms/min_jobs/max_jobs/rate_factor/scenario/burst_active/total_generated/jobs_per_second).
  • POST /api/engine/synth/control { action, value?, scenario? } — pause / resume / trigger_burst / set_interval / set_rate / set_scenario. 서버측 clamp (interval 10..=5000ms, rate 0.1..=1.0).

검증¶

• cargo build clean.
• cargo clippy --lib Phase 3 신규 코드 경고 0.
• cargo test --lib 85/85 통과 (engine 21건, synth 7건 추가):
  • engine::synth::scenario_parse_round_trip
  • engine::synth::pick_weighted_only_returns_eligible_types
  • engine::synth::rate_factor_clamps_and_scales_max
  • engine::synth::set_tick_ms_clamps
  • engine::synth::pause_resume_toggle_enabled
  • engine::synth::tick_pushes_into_fallback_queue_when_no_nodes
  • engine::synth::trigger_burst_marks_active

의도적 단순화 (이후 phase 에서 보강)¶

• process_rate 는 scheduler.snapshot().total_completed 의 monotonic 값만 참조 — 정확한 1초 슬라이딩 윈도우 throughput 은 Phase 8 확장에서 반영.
• getActiveNodes 콜백은 단순히 stream_manager.connected_count() + 1 (edge) — 실제 ACTIVE 상태 필터는 Phase 6 (activeness) 에서.
• Dev Console UI 컨트롤 버튼은 Phase 8.

Phase 5 완료 — Job 영속화 + AGX 정산¶

commit:server|bd7f78b

엔진의 데이터 루프 마감. 두 완료 경로 (실제 연결 노드 + edge 시뮬) 가 모두 job_history / job_receipts / aggregated_earnings 에 기록되고, dispatcher 가 완료 노드에 BalanceUpdateMsg 를 push.

변경¶

• src/engine/reward.rs (신규) — legacy AGX 공식 1:1 포팅. score_multiplier (4 buckets: 90+/70+/50+/<50 → 1.2/1.0/0.8/0.5), calculate_reward(base × complexity × resource_weight × score_mult), breakdown_ratios (job_type 별 cpu/network/storage 비율).
• src/engine/dispatcher.rs — db: PgPool 필드 추가, Assignment 에 assigned_user_id + assigned_node_score 캐시.
  • dispatch → try_assign 성공 시 job_history ASSIGNED row insert.
  • handle_job_complete → 보상 계산 → job_receipts insert + aggregated_earnings.credit (daily upsert) + job_history COMPLETED 갱신 + BalanceUpdateMsg push (현재 잔액/누적 수입/마지막 보상).
  • handle_job_failed → reward=0 receipt insert + failed_jobs 카운터 증가 + job_history FAILED 갱신.
  • 모든 영속화 best-effort: 실패는 WARN 로그만, bidi stream 은 끊지 않음 (graceful degradation).
• src/engine/scheduler.rs — db: PgPool 필드 + tick 비동기화.
  • roll_outcomes (sync 헬퍼) 로 !Send ThreadRng 가 await 경계를 넘지 않도록 격리.
  • tick 당 단일 aggregated_earnings.credit upsert (집계 단위).
  • tick 당 sample receipt 1건 (completed) + 1건 (failed) — legacy createSampleReceipt 패턴, DB 압력 제한.
• src/engine/edge_node.rs — EdgeNodeInfo.user_id 필드 추가, derive_edge_user_id (UUID v5 namespace + wallet, 결정적), ensure_edge_node 가 accounts.ensure_edge 도 호출.
• src/domain/node/stream.rs — ConnectedNode 에 owner_user_id: Option<String> + score: i32 필드 추가, register 시그니처 확장.
• src/handler/node.rs — connect_node 에서 register 호출 시 owner_user_id / score 전달.
• store helpers:
  • accounts::ensure_edge(pool, user_id, display_name) — synthetic edge account upsert.
  • nodes::ensure_edge — owner_user_id 인자 추가.
  • jobs::insert_assigned, jobs::mark_completed, jobs::mark_failed.
  • earnings::credit(pool, user_id, period_date, completed, failed, total/cpu/network/storage_agx) — UNIQUE (user_id, period_date, period_type) 기반 누적 upsert.
  • receipts::insert_with_status (status 명시 변형) + 기존 insert 는 reward>0 으로 status 자동 결정.
• Cargo.toml — uuid features 에 v5 추가.

Edge 합성 계정¶

EDGE_USER_NAMESPACE (고정 UUID v4 한 번 생성 후 코드에 pin) + edge wallet → UUID v5 결정적 도출. accounts 행 + nodes.owner_user_id 모두 이 UUID 로 참조 → edge 의 receipts/earnings 가 일반 사용자와 동일 query 경로로 노출.

검증¶

• cargo build clean.
• cargo clippy --lib Phase 5 신규 코드 경고 0.
• cargo test --lib 88/88 통과 (engine 24건, 5건 추가):
  • engine::reward::score_multiplier_buckets
  • engine::reward::calculate_reward_matches_legacy_compute
  • engine::reward::breakdown_ratios_sum_to_one
  • engine::edge_node::derive_edge_user_id_is_stable_v5
  • engine::scheduler::roll_outcomes_high_tier_mostly_completes

기존 dispatcher/scheduler/synth 테스트는 stub PgPool(acquire_timeout=50ms) 로 영속화 실패가 빠르게 swallow 되도록 갱신 — 본래 30s default 였으면 테스트 90s 소요했음.

DB 변경¶

마이그레이션 추가 없음. 기존 job_history / job_receipts / aggregated_earnings / accounts / nodes 스키마에 INSERT/UPDATE 만 함. (그래서 본 커밋의 refs 는 #254 미포함.)

다음 phase 로 미룬 항목¶

• Phase 6: 멀티 노드 redundancy (FindNodesForJob), timeout sweep (deadline 기반 reassign/drop), activeness penalty.
• Phase 8: Dev Console UI 의 Engine 탭 (snapshot + 컨트롤 버튼).

Phase 6 완료 — Dispatcher timeout sweep + 취소¶

commit:server|0dbdd8a

할당된 노드가 응답하지 않는 경우의 라이프사이클 마감. 5초 주기 sweep 으로 deadline 초과 assignment 를 정리하고, 노드에 JobCancelMsg 송신 + job_history.status='TIMED_OUT' 기록.

변경¶

• src/engine/dispatcher.rs:
  • total_timed_out: AtomicI64 카운터 + total_timed_out() 접근자.
  • spawn_sweep() — 백그라운드 tokio task 시작 (5초 tick, MissedTickBehavior::Skip).
  • sweep_timeouts() — 한 번의 sweep 실행. 만료 key 를 먼저 snapshot 한 후 (DashMap 순회 중 mutate 방지) 각 entry 처리:
    1. assignments 에서 제거
    2. JobCancelMsg { job_id, reason: "expired" } 송신 (할당 노드의 슬롯 해제 유도)
    3. jobs::mark_timed_out (best-effort, 실패 시 WARN)
    4. total_timed_out 증가 + WARN 로그
• src/store/jobs.rs::mark_timed_out — UPDATE job_history SET status='TIMED_OUT', completed_at=NOW() WHERE job_id = $1. row 없으면 no-op.
• src/server/mod.rs + src/main.rs — 부팅 시 dispatcher.clone().spawn_sweep() 호출.
• src/admin/mod.rs::api_engine_snapshot — 응답에 total_timed_out 추가.

동작 안전성¶

• 영속화 / 스트림 송신 모두 best-effort. 실패는 WARN 로그만, sweep loop 자체는 끊지 않음 (graceful degradation).
• DashMap 순회 race: 만료 key 를 먼저 collect → 그 다음 remove 패턴으로 안전.
• legacy 와 동일하게 자동 reassignment 없음 — 만료 job 은 drop. operator 는 admin snapshot 의 total_timed_out 증가로 인지하여 새로 enqueue 가능.

Phase 6 의도적 단순화 (issue scope 그대로)¶

• Activeness penalty 없음 (StreamManager 의 score 감산 / INACTIVE 자동 전이는 별도 enhancement).
• Multi-node redundancy (FindNodesForJob) 없음 (single-node 할당 유지).
• 둘 다 MonitorService 포팅 / 운영 데이터 수집 후 별도 이슈로 분리하는 게 합리적.

검증¶

• cargo build clean.
• cargo clippy --lib Phase 6 신규 코드 경고 0.
• cargo test --lib 89/89 통과 (engine 25건, 1건 추가):
  • engine::dispatcher::sweep_expires_overdue_assignments_and_keeps_fresh_ones — 만료/유효 assignment 혼합 → sweep 이 만료만 제거, fresh 유지, JobCancelMsg (reason=expired) 송신, total_timed_out=1.

DB 변경¶

마이그레이션 없음. 기존 job_history.status 컬럼 UPDATE 만. (#254 refs 불필요.)

Phase 7 완료 — VirtualNodeCreditor + virtual node 테이블¶

commit:server|d4d49f9

가상 노드 보유자에 대한 passive AGX 적립 루프 착륙. 1분 tick 으로 active assignment sweep → 만료된 행 expire 처리 + 활성 행에 대해 virtual_node_credits row 작성 (일자별 cap 적용).

변경¶

• migrations/0013_virtual_nodes.sql (신규) — 3 테이블, account-scope (user_id keyed):
  • virtual_node_packages — 구매 가능한 패키지 카탈로그 (informational, FK 없는 nullable 참조).
  • virtual_node_assignments — user 별 grant. status ∈ {active, expired, revoked}. 인덱스 (user_id), (status).
  • virtual_node_credits — period 별 적립 기록. 인덱스 (user_id, created_at DESC), (assignment_id, period_end DESC).
• src/store/virtual_nodes.rs (신규) — VirtualNodeAssignment struct + 5개 helper:
  • list_active, update_status, last_credit_period_end, insert_credit, total_credited_agx.
• src/engine/virtual_node_creditor.rs (신규) — 60초 tick 백그라운드 task. 각 active assignment 처리:
  • end_time < now → update_status('expired'), skip.
  • period_start = COALESCE(last_credit_period_end, assignment.start_time) → cursor.
  • period_end = min(now, end_time or now).
  • jobs_in_window 으로 일자별 cap 적용 (per-day cap = daily_job_cap × node_count) → 누적 jobs.
  • agx = jobs × reward_per_job (6 decimal round) → insert_credit.
• src/store/wallets.rs::get_balance — virtual_node_credits.agx_earned 도 합산하도록 확장 → total_earned_agx = receipts + virtual_credits. 별도 회계 경로 없이 잔액에 자동 반영.
• src/server/mod.rs + src/main.rs — 부팅 시 VirtualNodeCreditor::spawn(). AppState.creditor 추가.
• src/admin/mod.rs::api_engine_snapshot — 응답에 creditor 섹션 (tick_ms / last_tick_unix / total_credits_issued / total_assignments_expired / total_agx_credited).

검증¶

• cargo build clean.
• cargo clippy --lib Phase 7 신규 코드 경고 0.
• cargo test --lib 94/94 통과 (engine 30건, virtual_node_creditor 5건 추가):
  • jobs_in_window_below_daily_cap — 60s × 10 jps × 2 nodes = 1200 jobs (cap 미적용).
  • jobs_in_window_clamped_to_daily_cap — 30시간 누적 시 일자별 cap 200k 적용 → 400k.
  • jobs_in_window_zero_for_empty_window.
  • jobs_in_window_respects_assignment_end — 60s 요청해도 30s 에서 cut.
  • round_agx_six_decimals.

Phase 7 의도적 단순화¶

• AGX halving 미구현 — legacy halvedVirtualAgxEarnedForAssignmentWindow 는 별도 agx_halving_events 테이블 필요. Monitor/Agx 서비스 포팅 시 함께.
• Redeem code → assignment 자동 발급 연동 없음 — Redeem(#420) ↔ VNode 의 cross-cutting 별도 enhancement.
• assignment 생성/관리 admin REST 미구현 — 데모용 직접 SQL insert 또는 별도 admin 이슈.

다음¶

마지막 phase — Phase 8: Engine 통합 + admin 컨트롤 + Dev Console UI.