Jon Chery
df58bc25a3
---ci--- project: orca phase: 3 milestone: v0.3 status: complete requirements: covered: [REQ-022, REQ-030, REQ-032] partial: [] ---/ci--- v0.3 milestone merged to main. Includes all v0.2 work (P08-P10) that was previously on the milestone branch but not yet merged to main, plus the v0.3 completion work (iter.Seq streaming + doctor network/db). v0.2 phases included: P08 (mTLS), P09 (scheduling), P10 (security scan). v0.3 phases: P0 (pre-execution), P1 (iter.Seq streaming), P2 (doctor), P3 (final review+ship). Total: 40 requirements, all complete. No new go.mod dependencies. Full test suite passes under -race. gofmt + go vet clean.
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()
|
|
}
|