fc6a6c07e2
Wave A of P02 (multi-node scheduling & job dispatch).
- internal/store/migrations/0005_node_capacity.sql — node_capacity
table (node_id PK, cpu_millicores, memory_mib, disk_mib, updated_at).
- internal/store/capacity_repo.go — CRUD for the table; ErrNotFound
semantics; List ordered by node_id.
- internal/store/capacity_repo_test.go — round-trip coverage.
- internal/engine/peer.go — Peer struct (NodeID, Address, ServerName,
CAPath, LastSeen, Capacity) and PeerRegistry (in-memory map with
sync.RWMutex; Add/Remove/Get/All/Len/UpdateLastSeen). All() returns
a stable-sorted snapshot for deterministic tests.
- internal/engine/scheduler.go — JobSpec {CPU, Mem, Disk}; Fits()
and Score() helpers; PickNode() does best-fit bin-packing with
deterministic tie-breaking by NodeID. Ties broken lexicographically.
- internal/engine/scheduler_test.go — best-fit, no-fit, tie-break,
and Fits() boundary coverage.
- internal/transport/idempotency.go — IdempotencyStore (in-memory,
TTL=5min); WithIdempotencyKey/IdempotencyKeyFromContext helpers.
Expired entries auto-evict on Get; Sweep() for bulk cleanup.
- internal/transport/idempotency_test.go — put/get, expiry, ctx.
- internal/transport/retry.go — RetryPolicy (100ms/5s/5attempts);
IsTransient() with explicit signature list (no net/error dep);
ErrTransient/ErrPermanent sentinels; Do[T] generic retry loop.
Auto-retry only when (verb is idempotent) OR (ctx has idempotency
key); otherwise transient errors bail on first attempt (REQ-037).
backoff() with 25% jitter, ctx cancellation respected.
---ci---
project: orca
phase: 9
milestone: v0.2
status: execute
---/ci---
152 lines
4.4 KiB
Go
152 lines
4.4 KiB
Go
// Package transport — retry.go implements exponential backoff with
|
|
// jitter for cross-node dispatch retries. Per the P02 plan: 100ms
|
|
// initial, x2, 5s cap, max 5 attempts. Auto-retry only when the call
|
|
// is idempotent (X-Orca-Idempotency-Key header present, or the verb
|
|
// is intrinsically idempotent like GET/HEAD).
|
|
package transport
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"math/rand"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
// RetryInitial is the first backoff interval.
|
|
RetryInitial = 100 * time.Millisecond
|
|
// RetryMax is the cap on backoff between attempts.
|
|
RetryMax = 5 * time.Second
|
|
// RetryMaxAttempts is the total attempt count (including the first).
|
|
RetryMaxAttempts = 5
|
|
)
|
|
|
|
// RetryPolicy carries the backoff configuration. Zero value is the
|
|
// default (100ms / 5s / 5 attempts).
|
|
type RetryPolicy struct {
|
|
Initial time.Duration
|
|
Max time.Duration
|
|
MaxAttempts int
|
|
}
|
|
|
|
// DefaultRetryPolicy returns the P02 default.
|
|
func DefaultRetryPolicy() RetryPolicy {
|
|
return RetryPolicy{Initial: RetryInitial, Max: RetryMax, MaxAttempts: RetryMaxAttempts}
|
|
}
|
|
|
|
// IsTransient reports whether err looks like a transient failure
|
|
// worth retrying. We treat network errors, context-deadline-exceeded
|
|
// (peer was slow but reachable), and a sentinel ErrTransient as
|
|
// retryable; everything else (4xx, validation, auth) is permanent.
|
|
func IsTransient(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
if errors.Is(err, ErrTransient) {
|
|
return true
|
|
}
|
|
// We avoid pulling net/error here to keep dependencies minimal;
|
|
// the most common transient signature is the substring "connection
|
|
// refused" or "i/o timeout". Tests assert these explicitly.
|
|
s := err.Error()
|
|
for _, sub := range []string{"connection refused", "i/o timeout", "EOF", "no such host", "connection reset"} {
|
|
if contains(s, sub) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// ErrTransient is a sentinel callers can wrap to mark an error
|
|
// retryable. ErrPermanent is the opposite.
|
|
var (
|
|
ErrTransient = errors.New("transient error")
|
|
ErrPermanent = errors.New("permanent error")
|
|
)
|
|
|
|
// RetryableFunc is the signature Retry calls. It returns the result
|
|
// and an error. The bool indicates whether the call is idempotent
|
|
// (true = safe to retry without an idempotency key).
|
|
type RetryableFunc[T any] func(ctx context.Context, attempt int) (T, bool, error)
|
|
|
|
// Do runs fn with backoff according to policy. It retries only if
|
|
// (a) the call is idempotent, OR (b) ctx carries an idempotency key
|
|
// (set via WithIdempotencyKey). Otherwise a transient error on the
|
|
// first attempt is returned immediately (REQ-037: no retry without
|
|
// the key).
|
|
//
|
|
// The generic result T lets callers reuse this for jobIDs, status
|
|
// responses, etc. without boxing through `any`.
|
|
func Do[T any](ctx context.Context, p RetryPolicy, fn RetryableFunc[T]) (T, error) {
|
|
var zero T
|
|
if p.MaxAttempts <= 0 {
|
|
p = DefaultRetryPolicy()
|
|
}
|
|
hasKey := IdempotencyKeyFromContext(ctx) != ""
|
|
for attempt := 1; attempt <= p.MaxAttempts; attempt++ {
|
|
if err := ctx.Err(); err != nil {
|
|
return zero, err
|
|
}
|
|
v, idempotent, err := fn(ctx, attempt)
|
|
if err == nil {
|
|
return v, nil
|
|
}
|
|
// Permanent errors never retry.
|
|
if errors.Is(err, ErrPermanent) {
|
|
return zero, err
|
|
}
|
|
// Last attempt — surface the error.
|
|
if attempt == p.MaxAttempts {
|
|
return zero, err
|
|
}
|
|
// Transient + no idempotency + not idempotent verb: no retry.
|
|
if IsTransient(err) && !idempotent && !hasKey {
|
|
return zero, err
|
|
}
|
|
// Wait with jittered backoff, but respect ctx cancellation.
|
|
wait := backoff(p.Initial, p.Max, attempt)
|
|
t := time.NewTimer(wait)
|
|
select {
|
|
case <-ctx.Done():
|
|
t.Stop()
|
|
return zero, ctx.Err()
|
|
case <-t.C:
|
|
}
|
|
}
|
|
return zero, errors.New("retry.Do: exhausted attempts without error (impossible)")
|
|
}
|
|
|
|
// backoff returns the wait duration for the n-th attempt (1-indexed).
|
|
// Formula: min(Initial * 2^(n-1), Max), with up to 25% jitter.
|
|
func backoff(initial, max time.Duration, n int) time.Duration {
|
|
d := initial
|
|
for i := 1; i < n; i++ {
|
|
d *= 2
|
|
if d > max {
|
|
d = max
|
|
break
|
|
}
|
|
}
|
|
// Jitter: ±25% of d.
|
|
jitter := time.Duration(rand.Int63n(int64(d) / 2))
|
|
d = d - d/4 + jitter
|
|
if d < 0 {
|
|
d = 0
|
|
}
|
|
return d
|
|
}
|
|
|
|
// contains is a tiny substring helper (avoids pulling strings for one
|
|
// call site; this is hot-path retry classification).
|
|
func contains(s, sub string) bool {
|
|
if len(sub) == 0 {
|
|
return true
|
|
}
|
|
for i := 0; i+len(sub) <= len(s); i++ {
|
|
if s[i:i+len(sub)] == sub {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|