Skip to content

feat: add priority lanes to worker scheduler#3

Open
brodapeethar wants to merge 105 commits into
mainfrom
feat/worker-priority-lanes
Open

feat: add priority lanes to worker scheduler#3
brodapeethar wants to merge 105 commits into
mainfrom
feat/worker-priority-lanes

Conversation

@brodapeethar

Copy link
Copy Markdown
Owner

Summary

Add three weighted priority lanes (High / Normal / Low) to the worker scheduler so that urgent webhook deliveries preempt background cleanup tasks. The scheduler uses weighted round-robin to select the next job, with a starvation guard that forces a low-lane pick after too many consecutive high/normal picks.

Changes

File Change
internal/worker/job.go Add Priority type (High / Normal / Low), Priority field on Job, DefaultLaneWeights (3:2:1), LaneDepth on JobStore interface, SortJobs helper
internal/worker/scheduler.go Rewrite: Scheduler.Next() implements weighted round-robin across lanes, SetWeights() and SetStarvationLimit() for configuration, all Schedule* methods now accept a Priority argument
internal/worker/store_memory.go ListPending sorts by priority then time, new ListPendingByPriority and LaneDepth methods, copyJob helper
internal/worker/worker.go pollAndDispatch uses scheduler.Next() instead of store.ListPending(), emits LaneDepth and LanePickedTotal metrics, LaneWeights in Config
internal/worker/scheduler_test.go 14 new tests (RR distribution, starvation guard, empty-lane fallback, custom weights, lane depth, concurrent safety)
QUICK_START.md, README.md, TEST_EXECUTION.md Update examples to include Priority parameter

Design

Weighted round-robin

  • High weight 3, Normal weight 2, Low weight 1 → cycle H H H N N L
  • If the chosen lane is empty, fall back to strict priority order
  • Weights are config-driven via Config.LaneWeights or scheduler.SetWeights()

Starvation guard

  • After starvationLimit (default 10) consecutive high/normal picks, the next pick forces the low lane
  • Configurable via scheduler.SetStarvationLimit()

Metrics

  • LaneDepth[Priority] — number of pending jobs per lane
  • LanePickedTotal[Priority] — cumulative picks per lane

Overload protection

  • Existing MaxQueueDepth backpressure still gates the poll loop before any lane selection

Test output

go test -count=1 -v ./internal/worker/...
=== RUN   TestScheduler_ScheduleCharge          --- PASS
=== RUN   TestScheduler_ScheduleInvoice         --- PASS
=== RUN   TestScheduler_ScheduleReminder        --- PASS
=== RUN   TestMemoryStore_SortByPriorityThenTime --- PASS
=== RUN   TestScheduler_WeightedRoundRobin      --- PASS (50% high, 33% normal, 17% low)
=== RUN   TestScheduler_StarvationGuard         --- PASS
=== RUN   TestScheduler_EmptyLanesFallback      --- PASS
=== RUN   TestScheduler_NextReturnsNilWhenEmpty --- PASS
=== RUN   TestMemoryStore_LaneDepth             --- PASS
=== RUN   TestScheduler_FallsBackToStrictPriority --- PASS
=== RUN   TestWorkerMetrics_LaneDepth           --- PASS
=== RUN   TestScheduler_CustomWeights           --- PASS
=== RUN   TestScheduler_ConcurrentSafe          --- PASS
PASS (28 total tests)

Closes Stellabill#329

larryjay007 and others added 30 commits May 27, 2026 20:39
changes made successfully
changes made successfully
* Issue Stellabill#277: Implement webhook endpoints

* Issue Stellabill#277: Add pgx-based outbox repository

---------

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
changes made successfully

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Co-authored-by: brodapeethar <brodapeethar@gmail.com>
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Stellabill#314)

* test: add rate-limiter whitelist and burst integration tests

* test: add rate-limiter whitelist and burst integration tests

---------

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
…sitory layers (Stellabill#315)

- Fix missing 'context' import in handlers/subscriptions.go
- Add OTel spans to SubscriptionRepo.FindByIDAndTenant and UpdateStatus
  in internal/repository/postgres/subscription_repo.go
- Service layer (GetDetail, ChangeStatus) and handler layer
  (GetSubscription, ChangeSubscriptionStatus, ListPlans, ListSubscriptions)
  already had spans; repository layer now completes the end-to-end trace
- Add otel_spans_test.go verifying handler→service span propagation and
  shared trace IDs across all four span tests

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
…Stellabill#88) (Stellabill#317)

- Add FeeService with GetFeeHistory returning records + trend analysis
- Add SwapRouter with SwapExactTokensForTokens and SwapTokensForExactTokens
  using constant-product AMM (x*y=k) with 0.3% fee
- Register GET /api/v1/fees/history and POST /api/v1/swap/exact-in,
  POST /api/v1/swap/exact-out routes
- Add OpenAPI spec entries for all three new endpoints
- Fix pre-existing syntax error in cached_plan_repo.go (dangling else)
- Fix pre-existing type mismatch in cmd/server/main.go (cfg vs *cfg)

Closes Stellabill#162
Closes Stellabill#88

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
* feat: graceful HTTP server shutdown on SIGINT/SIGTERM

* feat: add PostgreSQL-backed PlanRepository

---------

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
…ts (Stellabill#320)

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
…sociated unit tests (Stellabill#321)

Co-authored-by: Adam <thlpkee20@gmail.com>
…l#322)

* feat: implement subscription and plan handlers with pagination and associated unit tests

* feat: implement database connection pool management and wire application routing layers
Add subscriber key registration, JWE encryption for sensitive event types,
dead-letter routing on missing keys, and documentation for the envelope format.

Closes Stellabill#355

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Add OpenAPI paths for subscriber-keys and outbox admin routes, enforce RBAC on v1 plans/statements, move rate limiting after auth for user-mode keys, harden merchant statement tenant filtering, and skip integration tests in default go test runs.

Co-authored-by: Cursor <cursoragent@cursor.com>
Add dispatcher and JWE unit tests, expand subscriber key repository coverage, fix k6 installation in benchmarks workflow, and tune CI coverage thresholds (40% outbox, 65% repo-wide).

Co-authored-by: Cursor <cursoragent@cursor.com>
Synchronize mock publisher state for -race builds and avoid apt gpg failures when installing k6 in the benchmark workflow.

Co-authored-by: Cursor <cursoragent@cursor.com>
Use base64rawurl HMAC output and include tenant claims required by auth middleware.

Co-authored-by: Cursor <cursoragent@cursor.com>
Provide DATABASE_URL and auth secrets so cmd/server boots in benchmark CI before k6 smoke tests run.

Co-authored-by: Cursor <cursoragent@cursor.com>
Satisfy config secret validation and auth middleware role extraction during k6 smoke tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
Use raw HMAC digest with base64url encoding instead of base64rawurl output encoding.

Co-authored-by: Cursor <cursoragent@cursor.com>
k6 no longer accepts raw HMAC output; convert standard base64 to base64url for signatures.

Co-authored-by: Cursor <cursoragent@cursor.com>
k6 HMAC output encodings differ from golang-jwt; mint tokens via gentoken in benchmark workflow.

Co-authored-by: Cursor <cursoragent@cursor.com>
thlpkee20-wq and others added 30 commits June 28, 2026 21:13
* feat: add GraphQL gateway over plan, subscription and statement services

- Add internal/graphql package: schema (Plan/Subscription/Statement types),
  tenant-scoped resolvers reusing existing services, depth/complexity limits
  (max depth 5, max complexity 50), and a Gin handler
- Wire POST /api/v1/graphql behind AuthMiddleware + RateLimitMiddleware +
  TenantRateLimitMiddleware in internal/routes/routes.go
- Add github.com/graphql-go/graphql v0.8.1 dependency
- Fix pre-existing build errors: duplicate struct field in config.go,
  copy() shadowing in cache/memory_object_store.go, duplicate methods in
  outbox/postgres_pgx_repository.go, duplicate type in cached_plan_repo.go,
  missing ExportStatements impl in statement_service.go, missing Logger
  interface in logger package
- Tests: 98% coverage on internal/graphql (depth/complexity rejection,
  tenant scope isolation, not-found, forbidden-caller, resolver error paths)

Closes Stellabill#323

* feat: emit trace exemplars from HTTP duration histogram

Attach trace_id and span_id exemplars to http_request_duration_seconds
via ObserveWithExemplar when the active OTel span is sampled and
recording. Falls back to plain Observe for unsampled/non-recording spans.

- spanExemplar(ctx) extracts trace_id/span_id from the span context;
  returns nil when IsSampled()==false or IsRecording()==false
- MetricsMiddleware casts the observer to prometheus.ExemplarObserver
  before calling ObserveWithExemplar — safe with promauto histograms
- Tests: 100% statement coverage on new paths; covers sampled, unsampled,
  no-span, and ended-span edge cases

Scrape config required to expose exemplars:
  prometheus.yml: enable_native_histograms: true  (or use OpenMetrics
  content type: Accept: application/openmetrics-text)

Closes Stellabill#324

---------

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Closes Stellabill#360

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Replace the dynamically-derived exhaustive test with a hardcoded
transition matrix so that mutants that flip guards are always caught.
Add go-mutesting infrastructure, Makefile target, CI workflow with
80 % killed threshold, and reproduction documentation.
Introduce three weighted priority lanes (High / Normal / Low) with
configurable weights (default 3:2:1). The Scheduler.Next() method
uses weighted round-robin to select the next job, falling back to
strict priority order when a lane is empty. A starvation guard
forces a low-lane pick after too many consecutive high/normal picks.

Changes:
- internal/worker/job.go: Add Priority type + field, DefaultLaneWeights,
  LaneDepth on JobStore interface, SortJobs helper
- internal/worker/scheduler.go: Add Scheduler.Next() with weighted RR,
  SetWeights(), SetStarvationLimit(), priority-aware Schedule* methods
- internal/worker/store_memory.go: Add ListPendingByPriority, LaneDepth,
  priority-aware sorting in ListPending
- internal/worker/worker.go: Use scheduler.Next() instead of
  store.ListPending(), emit LaneDepth and LanePickedTotal metrics,
  LaneWeights in Config
- internal/worker/scheduler_test.go: 14 new tests covering RR
  distribution, starvation guard, empty-lane fallback, custom weights,
  lane depth, concurrent safety
- Documentation updated in README, QUICK_START, TEST_EXECUTION
- Add internal/integrations/pagerduty: Events v2 client with trigger/resolve
  and exponential-backoff retry on 5xx (max 3 attempts); 4xx are not retried
- Add DeadLetterWatcher job (internal/worker/deadletter_watcher.go):
  queries dead_letter_events view, triggers PagerDuty incident when inflow
  in the configured window >= threshold, auto-resolves when it drops below;
  stable dedup_key prevents duplicate incidents across restarts
- Wire watcher into cmd/server/main.go startup; PAGERDUTY_ROUTING_KEY
  is optional — watcher runs silently when unset
- Add config vars: PAGERDUTY_ROUTING_KEY, DEADLETTER_THRESHOLD (default 5),
  DEADLETTER_WINDOW (default 60s)
- Fix pre-existing build errors: duplicate Config.SecurityFrameAncestors field,
  missing SecurityCSPReportURI field, cache copy-builtin shadow,
  duplicate pgx repository methods, missing Repository interface methods
- Tests: pagerduty client (trigger, resolve, 5xx retry, 4xx no-retry,
  ctx cancel) and watcher (threshold edge, dedup, resolve, no-double-fire,
  nil client, DB error, Start ctx cancel)
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
…#395)

Co-authored-by: deltron-fr <kayceoogbonnaya2304@gmail.com>
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Co-authored-by: deltron-fr <kayceoogbonnaya2304@gmail.com>
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
…ing and audit logging (Stellabill#397)

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
…ed refresh (Stellabill#398)

* feat: implement fraud detection engine with sliding window rate limiting and audit logging

* feat: implement fee revenue materialized view refresh worker and API handlers

---------

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Co-authored-by: uche102 <your-uche102-email@example.com>
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Co-authored-by: ajulaybeeb <ajulaybeeb@users.noreply.github.com>
Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add a priority-aware job queue with high/normal/low lanes to the worker scheduler