toki
348578e5dd
- Add edge server registry and connection management - Implement edge wire protocol with gRPC streaming - Update proto definitions for control plane communication - Add edge bootstrap runtime and connector - Add code review and plan documentation for G08
119 lines
3.6 KiB
Go
119 lines
3.6 KiB
Go
package wire
|
|
|
|
import (
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
proto_socket "git.toki-labs.com/toki/proto-socket/go"
|
|
iop "iop/proto/gen/iop"
|
|
)
|
|
|
|
// EdgeConnectionState captures the Control Plane's internal view of a single
|
|
// Edge wire connection. It is a registry foundation only; public status query
|
|
// and event relay APIs are deferred to later subtasks.
|
|
type EdgeConnectionState struct {
|
|
EdgeID string
|
|
EdgeName string
|
|
Version string
|
|
Capabilities []string
|
|
Protocol string
|
|
Connected bool
|
|
LastSeen time.Time
|
|
DisconnectInfo proto_socket.DisconnectInfo
|
|
|
|
// token identifies the specific connection that produced this state. It is an
|
|
// internal registry detail (not an external API contract) used to ignore a
|
|
// stale connection's late disconnect after the same edge_id reconnected.
|
|
token uint64
|
|
}
|
|
|
|
// EdgeRegistry stores Edge connection state keyed by edge_id behind a mutex.
|
|
type EdgeRegistry struct {
|
|
mu sync.Mutex
|
|
edges map[string]*EdgeConnectionState
|
|
nextToken uint64
|
|
}
|
|
|
|
// NewEdgeRegistry returns an empty Edge connection registry.
|
|
func NewEdgeRegistry() *EdgeRegistry {
|
|
return &EdgeRegistry{edges: make(map[string]*EdgeConnectionState)}
|
|
}
|
|
|
|
// MarkConnected records or refreshes an Edge as connected from its hello and
|
|
// returns the connection token to hand back on disconnect. A reconnection
|
|
// supersedes any earlier connection for the same edge_id.
|
|
func (r *EdgeRegistry) MarkConnected(req *iop.EdgeHelloRequest, at time.Time) uint64 {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.nextToken++
|
|
token := r.nextToken
|
|
r.edges[req.GetEdgeId()] = &EdgeConnectionState{
|
|
EdgeID: req.GetEdgeId(),
|
|
EdgeName: req.GetEdgeName(),
|
|
Version: req.GetVersion(),
|
|
Capabilities: append([]string(nil), req.GetCapabilities()...),
|
|
Protocol: Protocol,
|
|
Connected: true,
|
|
LastSeen: at,
|
|
token: token,
|
|
}
|
|
return token
|
|
}
|
|
|
|
// MarkDisconnected flips a tracked Edge to disconnected and records why, but
|
|
// only when token still matches the current connection. A stale token from a
|
|
// superseded connection is ignored so a reconnection is not cleared. It returns
|
|
// true only when the current connection was flipped, so callers (e.g. the
|
|
// EdgeServer active client map) can apply the same stale-safe cleanup.
|
|
func (r *EdgeRegistry) MarkDisconnected(edgeID string, token uint64, at time.Time, info proto_socket.DisconnectInfo) bool {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
state, ok := r.edges[edgeID]
|
|
if !ok || state.token != token {
|
|
return false
|
|
}
|
|
state.Connected = false
|
|
state.LastSeen = at
|
|
state.DisconnectInfo = info
|
|
return true
|
|
}
|
|
|
|
// Snapshot returns a copy of the tracked state for an Edge.
|
|
func (r *EdgeRegistry) Snapshot(edgeID string) (EdgeConnectionState, bool) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
state, ok := r.edges[edgeID]
|
|
if !ok {
|
|
return EdgeConnectionState{}, false
|
|
}
|
|
snapshot := *state
|
|
snapshot.Capabilities = append([]string(nil), state.Capabilities...)
|
|
return snapshot, true
|
|
}
|
|
|
|
// Snapshots returns a stable, copied view of all tracked Edge states.
|
|
func (r *EdgeRegistry) Snapshots() []EdgeConnectionState {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
ids := make([]string, 0, len(r.edges))
|
|
for id := range r.edges {
|
|
ids = append(ids, id)
|
|
}
|
|
sort.Strings(ids)
|
|
snapshots := make([]EdgeConnectionState, 0, len(ids))
|
|
for _, id := range ids {
|
|
state := r.edges[id]
|
|
snapshot := *state
|
|
snapshot.Capabilities = append([]string(nil), state.Capabilities...)
|
|
snapshots = append(snapshots, snapshot)
|
|
}
|
|
return snapshots
|
|
}
|
|
|
|
// Len returns the number of tracked Edge connections.
|
|
func (r *EdgeRegistry) Len() int {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return len(r.edges)
|
|
}
|