ciagent
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---
107 lines
2.7 KiB
Go
107 lines
2.7 KiB
Go
// Package engine — peer.go implements the peer registry for multi-node
|
|
// scheduling (v0.2 P02). A peer is a remote orca node reachable over
|
|
// mTLS. The registry is in-memory plus optionally SQLite-persisted;
|
|
// for P02 the in-memory map is the source of truth and persistence
|
|
// is best-effort.
|
|
package engine
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.cloudinit.dev/coreci/orca/internal/store"
|
|
)
|
|
|
|
// Peer is a remote orca node reachable over mTLS.
|
|
type Peer struct {
|
|
NodeID string
|
|
Address string // host:port (the peer's daemon listener)
|
|
ServerName string // expected SAN on the peer's cert
|
|
CAPath string // path to the CA cert this peer validates against
|
|
LastSeen time.Time
|
|
Capacity *store.NodeCapacity
|
|
}
|
|
|
|
// PeerRegistry tracks known peers. Methods are safe for concurrent
|
|
// use; the underlying map is guarded by a sync.RWMutex.
|
|
type PeerRegistry struct {
|
|
mu sync.RWMutex
|
|
peers map[string]*Peer
|
|
// optional persistence (not required for P02; can be added later)
|
|
persist PeerPersister
|
|
}
|
|
|
|
// PeerPersister is an optional callback for persisting peer records.
|
|
// P02 doesn't use it; it's here for the P03 audit log integration.
|
|
type PeerPersister interface {
|
|
SavePeer(ctx context.Context, p *Peer) error
|
|
}
|
|
|
|
// NewPeerRegistry returns an empty registry.
|
|
func NewPeerRegistry() *PeerRegistry {
|
|
return &PeerRegistry{peers: make(map[string]*Peer)}
|
|
}
|
|
|
|
// Add inserts or updates a peer record.
|
|
func (r *PeerRegistry) Add(p *Peer) error {
|
|
if p == nil {
|
|
return fmt.Errorf("PeerRegistry.Add: nil peer")
|
|
}
|
|
if p.NodeID == "" {
|
|
return fmt.Errorf("PeerRegistry.Add: NodeID is required")
|
|
}
|
|
r.mu.Lock()
|
|
r.peers[p.NodeID] = p
|
|
r.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
// Remove deletes a peer by ID. Returns true if a peer was removed.
|
|
func (r *PeerRegistry) Remove(nodeID string) bool {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
_, ok := r.peers[nodeID]
|
|
if ok {
|
|
delete(r.peers, nodeID)
|
|
}
|
|
return ok
|
|
}
|
|
|
|
// Get returns the peer with the given ID, or nil.
|
|
func (r *PeerRegistry) Get(nodeID string) *Peer {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
return r.peers[nodeID]
|
|
}
|
|
|
|
// All returns a snapshot of all peers, sorted by NodeID for determinism.
|
|
func (r *PeerRegistry) All(_ context.Context) ([]*Peer, error) {
|
|
r.mu.RLock()
|
|
out := make([]*Peer, 0, len(r.peers))
|
|
for _, p := range r.peers {
|
|
out = append(out, p)
|
|
}
|
|
r.mu.RUnlock()
|
|
sort.Slice(out, func(i, j int) bool { return out[i].NodeID < out[j].NodeID })
|
|
return out, nil
|
|
}
|
|
|
|
// Len returns the number of registered peers.
|
|
func (r *PeerRegistry) Len() int {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
return len(r.peers)
|
|
}
|
|
|
|
// UpdateLastSeen bumps the LastSeen timestamp on a peer.
|
|
func (r *PeerRegistry) UpdateLastSeen(nodeID string) {
|
|
r.mu.Lock()
|
|
if p, ok := r.peers[nodeID]; ok {
|
|
p.LastSeen = time.Now().UTC()
|
|
}
|
|
r.mu.Unlock()
|
|
}
|