Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion server/internal/checkworker/worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -1775,6 +1775,11 @@ func (r *CheckWorker) saveErrorResult(ctx context.Context, checkJob *models.Chec
// figure, so charging them a per-minute token would let server plumbing eat a
// customer's execution budget and tick their skipped-today counter.
//
// A job this gate already deferred may borrow from the bucket
// (ReserveJobExecution): claim priority alone cannot save a
// long-period check whose single attempt per period always lands just after
// its siblings drained the bucket.
//
// Returns deferred=true when the job was turned away (the caller must stop and
// return err); false means the check may run.
func (r *CheckWorker) applyRateLimitGate(
Expand All @@ -1789,7 +1794,7 @@ func (r *CheckWorker) applyRateLimitGate(
return false, nil
}

rateErr := entSvc.ReserveCheckExecution(ctx, checkJob.OrganizationUID)
rateErr := entSvc.ReserveJobExecution(ctx, checkJob)
if rateErr == nil {
return false, nil
}
Expand Down
12 changes: 12 additions & 0 deletions server/internal/db/models/check_job.go
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,18 @@ func (j *CheckJob) IsInternal() bool {
return j.Check != nil && j.Check.Internal
}

// RateLimitDeferred reports whether the per-org rate limiter turned this job
// away at its previous tick. A deferral advances scheduled_at by a period but
// keeps effective_scheduled_at (spec 2026-08-26-02), so the two end up about a
// period apart. A normal job has them within the scheduling offsets (seconds).
func (j *CheckJob) RateLimitDeferred() bool {
if j.ScheduledAt == nil || j.EffectiveScheduledAt == nil || j.Period <= 0 {
return false
}

return j.ScheduledAt.Sub(*j.EffectiveScheduledAt) >= time.Duration(j.Period)/2
}

// CheckJobUpdate represents fields that can be updated.
type CheckJobUpdate struct {
Region *string
Expand Down
50 changes: 50 additions & 0 deletions server/internal/db/models/check_job_deferred_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
package models_test

import (
"testing"
"time"

"github.com/stretchr/testify/require"

"github.com/fclairamb/solidping/server/internal/db/models"
"github.com/fclairamb/solidping/server/internal/utils/timeutils"
)

func TestCheckJobRateLimitDeferred(t *testing.T) {
t.Parallel()

scheduled := time.Date(2026, 10, 19, 8, 35, 46, 0, time.UTC)
at := func(d time.Duration) *time.Time {
v := scheduled.Add(d)

return &v
}

cases := []struct {
name string
period time.Duration
effective *time.Time
want bool
}{
{name: "on time", period: time.Hour, effective: at(0), want: false},
{name: "slow-check offset", period: time.Minute, effective: at(30 * time.Second), want: false},
{name: "paid tier credit", period: time.Hour, effective: at(-10 * time.Second), want: false},
{name: "deferred once", period: time.Hour, effective: at(-time.Hour), want: true},
{name: "deferred for a day", period: time.Hour, effective: at(-31 * time.Hour), want: true},
{name: "one-minute deferred with offset", period: time.Minute, effective: at(-30 * time.Second), want: true},
{name: "no effective schedule", period: time.Hour, effective: nil, want: false},
}

for _, testCase := range cases {
t.Run(testCase.name, func(t *testing.T) {
t.Parallel()

job := &models.CheckJob{
Period: timeutils.Duration(testCase.period),
ScheduledAt: &scheduled,
EffectiveScheduledAt: testCase.effective,
}
require.Equal(t, testCase.want, job.RateLimitDeferred())
})
}
}
217 changes: 217 additions & 0 deletions server/internal/entitlements/long_period_starvation_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,217 @@
package entitlements_test

import (
"sort"
"testing"
"time"

"github.com/stretchr/testify/require"

"github.com/fclairamb/solidping/server/internal/config"
"github.com/fclairamb/solidping/server/internal/db/models"
"github.com/fclairamb/solidping/server/internal/db/sqlite"
"github.com/fclairamb/solidping/server/internal/entitlements"
"github.com/fclairamb/solidping/server/internal/utils/timeutils"
)

// starvationJob is one region-job of the simulated org: its period and its
// fixed phase (the UID-hash offset within the period), in seconds.
type starvationJob struct {
name string
period time.Duration
phase time.Duration
long bool
}

// starvationOrg reproduces the prod org of 2026-10-18: 9 one-minute checks,
// 7 five-minute checks and 8 hourly checks, all in one region, against the
// SaaS cap of 10/min (demand ~10.5/min). The phases are the ones read from
// check_jobs. Two hourly checks land one second after a burst of one-minute
// siblings, so the bucket is empty at their only attempt of the hour.
func starvationOrg() []starvationJob {
sec := func(m, s int) time.Duration { return time.Duration(m)*time.Minute + time.Duration(s)*time.Second }

jobs := make([]starvationJob, 0, 24)
for i, s := range []int{0, 1, 14, 15, 26, 29, 39, 42, 45} {
jobs = append(jobs, starvationJob{name: "minute-" + string(rune('a'+i)), period: time.Minute, phase: sec(0, s)})
}

for i, p := range []time.Duration{sec(0, 54), sec(1, 11), sec(3, 39), sec(3, 34), sec(0, 8), sec(0, 40), sec(0, 48)} {
jobs = append(jobs, starvationJob{name: "five-" + string(rune('a'+i)), period: 5 * time.Minute, phase: p})
}

for i, p := range []time.Duration{
sec(35, 46), sec(6, 1), sec(44, 7), sec(2, 15), sec(0, 39), sec(26, 59), sec(18, 38), sec(40, 35),
} {
jobs = append(jobs, starvationJob{name: "hour-" + string(rune('a'+i)), period: time.Hour, phase: p, long: true})
}

return jobs
}

// simulateStarvation drives the real per-org token bucket second by second for
// `hours`, with the worker gate's bookkeeping: a run re-anchors
// effective_scheduled_at, a deferral advances scheduled_at by one period and
// keeps it (checkjobsvc.DeferLeaseRateLimited). Same-second arrivals are taken
// in claim order (effective_scheduled_at ascending). With borrow false every job
// reserves normally, which is the behavior before the fix; with borrow true it
// goes through ReserveJobExecution, the production gate. Returns executions per job.
func simulateStarvation(t *testing.T, hours int, borrow bool) (map[string]int, int) {
t.Helper()
r := require.New(t)
ctx := t.Context()

dbSvc, err := sqlite.New(ctx, sqlite.Config{InMemory: true})
r.NoError(err)
r.NoError(dbSvc.Initialize(ctx))
t.Cleanup(func() { _ = dbSvc.Close() })

org := models.NewOrganization("acme", "Acme")
r.NoError(dbSvc.CreateOrganization(ctx, org))

start := time.Date(2026, 10, 18, 0, 0, 0, 0, time.UTC)
now := start
svc := entitlements.NewService(
dbSvc, entitlements.DefaultsFor(config.DeploymentModeSaaS), 0,
entitlements.WithClock(func() time.Time { return now }),
)

specs := starvationOrg()
jobs := make([]*models.CheckJob, len(specs))

for i, spec := range specs {
at := start.Add(spec.phase)
jobs[i] = &models.CheckJob{
OrganizationUID: org.UID,
Period: timeutils.Duration(spec.period),
ScheduledAt: &at,
EffectiveScheduledAt: &at,
}
}

executions := make(map[string]int, len(specs))
total := 0
end := start.Add(time.Duration(hours) * time.Hour)

for ; now.Before(end); now = now.Add(time.Second) {
var due []int

for i, job := range jobs {
if job.ScheduledAt.Equal(now) {
due = append(due, i)
}
}

sort.SliceStable(due, func(a, b int) bool {
return jobs[due[a]].EffectiveScheduledAt.Before(*jobs[due[b]].EffectiveScheduledAt)
})

for _, i := range due {
job := jobs[i]

reserve := func() error { return svc.ReserveCheckExecution(ctx, org.UID) }
if borrow {
// The production gate.
reserve = func() error { return svc.ReserveJobExecution(ctx, job) }
}

next := job.ScheduledAt.Add(time.Duration(job.Period))
if reserve() == nil {
executions[specs[i].name]++
total++
job.EffectiveScheduledAt = &next
}

job.ScheduledAt = &next
}
}

return executions, total
}

// TestDeferredJobBorrowsSoLongPeriodChecksRun is the regression test for the
// prod watchdog report of 2026-10-19: two hourly checks of an org slightly
// over its cap had no result for 27 hours, while its one-minute checks shared
// the deficit fairly. The control runs the same simulation without borrowing
// and must reproduce the starvation, or the simulation proves nothing.
//
//nolint:paralleltest // two long simulations, each on its own in-memory DB
func TestDeferredJobBorrowsSoLongPeriodChecksRun(t *testing.T) {
const hours = 12

t.Run("ControlWithoutBorrowingStarvesAnHourlyCheck", func(t *testing.T) { //nolint:paralleltest // see above
executions, _ := simulateStarvation(t, hours, false)
t.Logf("executions without borrowing: %v", executions)

starved := 0

for _, spec := range starvationOrg() {
if spec.long && executions[spec.name] == 0 {
starved++
}
}

require.Positive(t, starved, "the control must reproduce a starved hourly check: %v", executions)
})

t.Run("BorrowingRunsEveryCheckAndKeepsTheCap", func(t *testing.T) { //nolint:paralleltest // see above
executions, total := simulateStarvation(t, hours, true)
t.Logf("executions with borrowing (total %d): %v", total, executions)

for _, spec := range starvationOrg() {
slots := int(time.Duration(hours) * time.Hour / spec.period)
require.GreaterOrEqual(t, executions[spec.name], slots/2,
"%s must run at least every other period: %v", spec.name, executions)
}

// SaaS cap 10/min: one burst at start plus at most one burst of debt.
const capPerMinute = 10
require.LessOrEqual(t, total, hours*60*capPerMinute+2*capPerMinute,
"borrowing must not lift the org's long-run rate above its cap")
})
}

// TestReserveDeferredCheckExecutionDebtIsBounded pins the borrowing contract: a
// deferred job may take the drained bucket down to one burst below zero, no
// further, and on-time jobs wait until refill has repaid the debt.
func TestReserveDeferredCheckExecutionDebtIsBounded(t *testing.T) {
t.Parallel()
r := require.New(t)
ctx := t.Context()

dbSvc, err := sqlite.New(ctx, sqlite.Config{InMemory: true})
r.NoError(err)
r.NoError(dbSvc.Initialize(ctx))
t.Cleanup(func() { _ = dbSvc.Close() })

org := models.NewOrganization("acme", "Acme")
r.NoError(dbSvc.CreateOrganization(ctx, org))

now := time.Date(2026, 10, 18, 0, 0, 0, 0, time.UTC)
svc := entitlements.NewService(
dbSvc, entitlements.DefaultsFor(config.DeploymentModeSaaS), 0,
entitlements.WithClock(func() time.Time { return now }),
)

// SaaS cap 10/min: drain the burst.
for i := range 10 {
r.NoError(svc.ReserveCheckExecution(ctx, org.UID), "call %d", i+1)
}

r.ErrorIs(svc.ReserveCheckExecution(ctx, org.UID), entitlements.ErrEntitlementExceeded)

for i := range 10 {
r.NoError(svc.ReserveDeferredCheckExecution(ctx, org.UID), "borrow %d", i+1)
}

r.ErrorIs(svc.ReserveDeferredCheckExecution(ctx, org.UID), entitlements.ErrEntitlementExceeded,
"the debt stops at one burst")

// One minute refills 10 tokens: the debt is repaid, the bucket is at 0.
now = now.Add(time.Minute)
r.ErrorIs(svc.ReserveCheckExecution(ctx, org.UID), entitlements.ErrEntitlementExceeded,
"on-time jobs wait until the debt is repaid")

now = now.Add(7 * time.Second)
r.NoError(svc.ReserveCheckExecution(ctx, org.UID), "refill after the debt serves on-time jobs again")
}
34 changes: 30 additions & 4 deletions server/internal/entitlements/service.go
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,31 @@ func (s *Service) CheckMembership(ctx context.Context, orgUID string) error {
//
// nil cap = unlimited (skip lookup, no allocation).
func (s *Service) ReserveCheckExecution(ctx context.Context, orgUID string) error {
return s.reserveCheckExecution(ctx, orgUID, false)
}

// ReserveDeferredCheckExecution is ReserveCheckExecution for a job the gate
// already turned away at least once (models.CheckJob.RateLimitDeferred). On a
// drained bucket it may borrow: the bucket goes into debt, down to one burst
// below zero, and the next on-time arrivals of the org are deferred instead.
//
// Without it a long-period check starves forever: its one attempt per period
// lands on the same UID-hash second, and when frequent siblings drain the
// bucket just before that second, the claim priority it earned is worth
// nothing (prod 2026-10-18, two hourly checks with zero results for 27 hours).
// The debt is repaid by refill, so the org's long-run rate stays at the cap.
func (s *Service) ReserveDeferredCheckExecution(ctx context.Context, orgUID string) error {
return s.reserveCheckExecution(ctx, orgUID, true)
}

// ReserveJobExecution is the rate gate of a claimed job: it borrows
// (ReserveDeferredCheckExecution) when the gate deferred the job last time,
// and reserves normally otherwise.
func (s *Service) ReserveJobExecution(ctx context.Context, job *models.CheckJob) error {
return s.reserveCheckExecution(ctx, job.OrganizationUID, job.RateLimitDeferred())
}

func (s *Service) reserveCheckExecution(ctx context.Context, orgUID string, borrow bool) error {
resolved, err := s.Resolve(ctx, orgUID)
if err != nil {
return fmt.Errorf("resolve entitlements: %w", err)
Expand All @@ -341,7 +366,7 @@ func (s *Service) ReserveCheckExecution(ctx context.Context, orgUID string) erro
}

bucket := s.limiterFor(orgUID, limit)
if !bucket.allow(s.now()) {
if !bucket.allow(s.now(), borrow) {
return &QuotaError{
LimitName: "MaxChecksPerMinute",
Limit: limit,
Expand Down Expand Up @@ -783,14 +808,15 @@ func (b *tokenBucket) refillLocked(now time.Time) {
}

// allow consumes one token if available. Returns false when the bucket
// is empty.
func (b *tokenBucket) allow(now time.Time) bool {
// is empty. With borrow, it may also take the bucket into debt, down to
// -capacity: the debt bounds the overshoot to one extra burst.
func (b *tokenBucket) allow(now time.Time, borrow bool) bool {
b.mu.Lock()
defer b.mu.Unlock()

b.refillLocked(now)

if b.tokens >= 1 {
if b.tokens >= 1 || (borrow && b.tokens > -float64(b.capacity)) {
b.tokens--

return true
Expand Down
2 changes: 1 addition & 1 deletion server/internal/handlers/agentws/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -1063,7 +1063,7 @@ func (h *Handler) handleClaim(
// MaxChecks slot and are absent from its demand figure, so they must
// not spend its per-minute budget or tick its skipped-today counter.
if h.entitlements != nil && !job.IsInternal() {
if rateErr := h.entitlements.ReserveCheckExecution(ctx, job.OrganizationUID); rateErr != nil {
if rateErr := h.entitlements.ReserveJobExecution(ctx, job); rateErr != nil {
var quotaErr *entitlements.QuotaError
if errors.As(rateErr, &quotaErr) {
// Same daily per-org tally as the in-process worker gate:
Expand Down
Loading