opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

style: fix golangci-lint formatting issues

Browse source 
- Run gofmt on all swarm package files
- Fix unnecessary int64 conversion in metrics.go
- Break long line in leader_election.go for golines

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Zhaoyikaiii 2026-02-23 22:08:12 +08:00
parent 050f99ef0c
commit d17bc02493
11 changed files with 156 additions and 155 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
pkg/swarm/config.go
View file

@ -209,13 +209,13 @@ func DefaultConfig() *Config {
BindAddr: "0.0.0.0",
BindPort: DefaultBindPort,
Discovery: DiscoveryConfig{
JoinAddrs: nil,
GossipInterval: Duration{DefaultGossipInterval},
PushPullInterval: Duration{DefaultPushPullInterval},
NodeTimeout: Duration{DefaultNodeTimeout},
DeadNodeTimeout: Duration{DefaultDeadNodeTimeout},
AuthSecret: "",
RequireAuth: false,
JoinAddrs: nil,
GossipInterval: Duration{DefaultGossipInterval},
PushPullInterval: Duration{DefaultPushPullInterval},
NodeTimeout: Duration{DefaultNodeTimeout},
DeadNodeTimeout: Duration{DefaultDeadNodeTimeout},
AuthSecret: "",
RequireAuth: false,
EnableMessageSigning: false,
},
Handoff: HandoffConfig{
@ -230,16 +230,16 @@ func DefaultConfig() *Config {
Timeout: Duration{10 * time.Second},
},
LoadMonitor: LoadMonitorConfig{
Enabled: true,
Interval: Duration{DefaultLoadSampleInterval},
SampleSize: DefaultLoadSampleSize,
CPUWeight: DefaultCPUWeight,
MemoryWeight: DefaultMemoryWeight,
SessionWeight: DefaultSessionWeight,
Enabled: true,
Interval: Duration{DefaultLoadSampleInterval},
SampleSize: DefaultLoadSampleSize,
CPUWeight: DefaultCPUWeight,
MemoryWeight: DefaultMemoryWeight,
SessionWeight: DefaultSessionWeight,
OffloadThreshold: DefaultOffloadThreshold,
MaxMemoryBytes: DefaultMaxMemoryBytes,
MaxGoroutines: DefaultMaxGoroutines,
MaxSessions: DefaultMaxSessions,
MaxMemoryBytes: DefaultMaxMemoryBytes,
MaxGoroutines: DefaultMaxGoroutines,
MaxSessions: DefaultMaxSessions,
},
LeaderElection: LeaderElectionConfig{
Enabled: false,
@ -247,9 +247,9 @@ func DefaultConfig() *Config {
LeaderHeartbeatTimeout: Duration{10 * time.Second},
},
Metrics: MetricsConfig{
Enabled: false,
ExportInterval: Duration{10 * time.Second},
PrometheusEnabled: false,
Enabled: false,
ExportInterval: Duration{10 * time.Second},
PrometheusEnabled: false,
PrometheusEndpoint: "/metrics",
},
}

20
pkg/swarm/discovery.go
View file

@ -31,10 +31,10 @@ type DiscoveryService struct {
rpcConn net.Listener
auth *AuthProvider
mu sync.RWMutex
running bool
stopChan chan struct{}
once sync.Once
mu sync.RWMutex
running bool
stopChan chan struct{}
once sync.Once
// Sequence number for updates
seqNum uint64
@ -281,13 +281,13 @@ func (ds *DiscoveryService) gossipLoop() {
// GossipMessage represents a gossip message.
type GossipMessage struct {
Type string `json:"type"` // "ping", "pong", "join", "update"
FromNode string `json:"from_node"`
SeqNum uint64 `json:"seq_num"`
Timestamp int64 `json:"timestamp"`
Payload []byte `json:"payload,omitempty"`
Type string `json:"type"` // "ping", "pong", "join", "update"
FromNode string `json:"from_node"`
SeqNum uint64 `json:"seq_num"`
Timestamp int64 `json:"timestamp"`
Payload []byte `json:"payload,omitempty"`
Nodes []*NodeInfo `json:"nodes,omitempty"` // For memberlist exchange
AuthToken *AuthToken `json:"auth_token,omitempty"`
AuthToken *AuthToken `json:"auth_token,omitempty"`
}
// handleGossip handles an incoming gossip message.

70
pkg/swarm/handoff.go
View file

@ -21,11 +21,11 @@ import (
type HandoffReason string
const (
ReasonOverloaded HandoffReason = "overloaded" // Load is too high
ReasonNoCapability HandoffReason = "no_capability" // Missing capability
ReasonUserRequest HandoffReason = "user_request" // User explicitly requested
ReasonNodeLeave HandoffReason = "node_leave" // Node is leaving
ReasonShutdown HandoffReason = "shutdown" // Graceful shutdown
ReasonOverloaded HandoffReason = "overloaded" // Load is too high
ReasonNoCapability HandoffReason = "no_capability" // Missing capability
ReasonUserRequest HandoffReason = "user_request" // User explicitly requested
ReasonNodeLeave HandoffReason = "node_leave" // Node is leaving
ReasonShutdown HandoffReason = "shutdown" // Graceful shutdown
)
// HandoffState represents the state of a handoff operation.
@ -42,27 +42,27 @@ const (
// HandoffRequest represents a request to hand off a session.
type HandoffRequest struct {
RequestID string `json:"request_id"`
Reason HandoffReason `json:"reason"`
SessionKey string `json:"session_key"`
SessionMessages []SessionMessage `json:"session_messages,omitempty"`
Context map[string]any `json:"context,omitempty"`
RequiredCap string `json:"required_cap,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
FromNodeID string `json:"from_node_id"`
FromNodeAddr string `json:"from_node_addr"`
Timestamp int64 `json:"timestamp"`
RequestID string `json:"request_id"`
Reason HandoffReason `json:"reason"`
SessionKey string `json:"session_key"`
SessionMessages []SessionMessage `json:"session_messages,omitempty"`
Context map[string]any `json:"context,omitempty"`
RequiredCap string `json:"required_cap,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
FromNodeID string `json:"from_node_id"`
FromNodeAddr string `json:"from_node_addr"`
Timestamp int64 `json:"timestamp"`
}
// HandoffResponse represents the response to a handoff request.
type HandoffResponse struct {
RequestID string `json:"request_id"`
Accepted bool `json:"accepted"`
NodeID string `json:"node_id"`
Reason string `json:"reason,omitempty"`
SessionKey string `json:"session_key,omitempty"` // New session key on target
Timestamp int64 `json:"timestamp"`
State HandoffState `json:"state"`
RequestID string `json:"request_id"`
Accepted bool `json:"accepted"`
NodeID string `json:"node_id"`
Reason string `json:"reason,omitempty"`
SessionKey string `json:"session_key,omitempty"` // New session key on target
Timestamp int64 `json:"timestamp"`
State HandoffState `json:"state"`
}
// HandoffCoordinator coordinates handoff operations between nodes.
@ -71,9 +71,9 @@ type HandoffCoordinator struct {
membership *MembershipManager
config HandoffConfig
pending map[string]*HandoffOperation // request_id -> operation
mu sync.RWMutex
conn *net.UDPConn
pending map[string]*HandoffOperation // request_id -> operation
mu sync.RWMutex
conn *net.UDPConn
// Accept/reject callbacks
onHandoffRequest func(*HandoffRequest) *HandoffResponse
@ -82,22 +82,22 @@ type HandoffCoordinator struct {
// HandoffOperation represents an ongoing handoff operation.
type HandoffOperation struct {
Request *HandoffRequest
Response *HandoffResponse
State HandoffState
StartTime time.Time
LastUpdate time.Time
RetryCount int
TargetNode *NodeWithState
Request *HandoffRequest
Response *HandoffResponse
State HandoffState
StartTime time.Time
LastUpdate time.Time
RetryCount int
TargetNode *NodeWithState
}
// NewHandoffCoordinator creates a new handoff coordinator.
func NewHandoffCoordinator(ds *DiscoveryService, config HandoffConfig) (*HandoffCoordinator, error) {
hc := &HandoffCoordinator{
discovery: ds,
discovery: ds,
membership: ds.membership,
config: config,
pending: make(map[string]*HandoffOperation),
config: config,
pending: make(map[string]*HandoffOperation),
}
// Bind UDP socket for handoff messages

3
pkg/swarm/leader_election.go
View file

@ -178,7 +178,8 @@ func (le *LeaderElection) monitorLeader() {
// Check if leader is still in the membership
if _, exists := le.membership.GetNode(leaderID); !exists {
logger.WarnCF("swarm", "Leader no longer in membership, triggering reelection", map[string]any{"leader_id": leaderID})
logger.WarnCF("swarm", "Leader no longer in membership, triggering reelection",
map[string]any{"leader_id": leaderID})
// Trigger reelection by clearing current leader
le.mu.Lock()
le.currentLeader = ""

32
pkg/swarm/load_monitor.go
View file

@ -14,9 +14,9 @@ import (
// LoadMonitor monitors system load and calculates a load score.
type LoadMonitor struct {
config *LoadMonitorConfig
samples []float64
mu sync.RWMutex
config *LoadMonitorConfig
samples []float64
mu sync.RWMutex
sessionCount int
ticker *time.Ticker
stopChan chan struct{}
@ -33,9 +33,9 @@ func NewLoadMonitor(config *LoadMonitorConfig) *LoadMonitor {
}
lm := &LoadMonitor{
config: config,
samples: make([]float64, 0, config.SampleSize),
stopChan: make(chan struct{}),
config: config,
samples: make([]float64, 0, config.SampleSize),
stopChan: make(chan struct{}),
onThreshold: make([]func(float64), 0),
}
return lm
@ -89,20 +89,20 @@ func (lm *LoadMonitor) run() {
// LoadMetrics represents current load metrics.
type LoadMetrics struct {
CPUUsage float64 `json:"cpu_usage"`
MemoryUsage float64 `json:"memory_usage"`
ActiveSessions int `json:"active_sessions"`
Goroutines int `json:"goroutines"`
Score float64 `json:"score"`
Timestamp int64 `json:"timestamp"`
CPUUsage float64 `json:"cpu_usage"`
MemoryUsage float64 `json:"memory_usage"`
ActiveSessions int `json:"active_sessions"`
Goroutines int `json:"goroutines"`
Score float64 `json:"score"`
Timestamp int64 `json:"timestamp"`
}
// GetCurrentLoad returns the current load metrics.
func (lm *LoadMonitor) GetCurrentLoad() *LoadMetrics {
metrics := &LoadMetrics{
ActiveSessions: lm.GetSessionCount(),
Goroutines: runtime.NumGoroutine(),
Timestamp: time.Now().UnixNano(),
Goroutines: runtime.NumGoroutine(),
Timestamp: time.Now().UnixNano(),
}
// Get memory usage
@ -245,7 +245,7 @@ func (lm *LoadMonitor) GetTrend() string {
// Simple linear regression to detect trend
n := float64(len(lm.samples))
sumX := n*(n-1)/2
sumX := n * (n - 1) / 2
sumY := 0.0
sumXY := 0.0
@ -255,7 +255,7 @@ func (lm *LoadMonitor) GetTrend() string {
sumXY += x * s
}
slope := (n*sumXY - sumX*sumY) / (n*(n-1)*(2*n-1)/6)
slope := (n*sumXY - sumX*sumY) / (n * (n - 1) * (2*n - 1) / 6)
if slope > TrendIncreasingThreshold {
return "increasing"

6
pkg/swarm/membership.go
View file

@ -20,9 +20,9 @@ type MembershipManager struct {
mu sync.RWMutex
// Event callbacks
onJoin []func(*NodeInfo)
onLeave []func(*NodeInfo)
onUpdate []func(*NodeInfo)
onJoin []func(*NodeInfo)
onLeave []func(*NodeInfo)
onUpdate []func(*NodeInfo)
}
// NewMembershipManager creates a new membership manager.

46
pkg/swarm/metrics.go
View file

@ -19,13 +19,13 @@ type MetricsCollector struct {
mu sync.RWMutex
// Counters (atomic for performance)
messagesSent atomic.Int64
messagesReceived atomic.Int64
messagesSent atomic.Int64
messagesReceived atomic.Int64
handoffsInitiated atomic.Int64
handoffsAccepted atomic.Int64
handoffsRejected atomic.Int64
handoffsFailed atomic.Int64
electionsWon atomic.Int64
handoffsAccepted atomic.Int64
handoffsRejected atomic.Int64
handoffsFailed atomic.Int64
electionsWon atomic.Int64
// Gauges (use atomic.Value for float64)
currentLoadScore atomic.Value // float64
@ -40,9 +40,9 @@ type MetricsCollector struct {
// LatencyBucket tracks latency distribution.
type LatencyBucket struct {
mu sync.RWMutex
count int64
sum int64
mu sync.RWMutex
count int64
sum int64
buckets [12]int64 // 0-1ms, 1-2ms, 2-5ms, 5-10ms, 10-20ms, 20-50ms, 50-100ms, 100-200ms, 200-500ms, 500ms-1s, 1-2s, 2s+
}
@ -118,7 +118,7 @@ func (m *MetricsCollector) RecordLatency(operation string, latency time.Duration
bucket := m.latencyBuckets[operation]
m.mu.Unlock()
ms := int64(latency.Milliseconds())
ms := latency.Milliseconds()
bucket.mu.Lock()
bucket.count++
@ -174,24 +174,24 @@ func (m *MetricsCollector) GetMetrics() map[string]any {
return map[string]any{
// Counters
"messages_sent": m.messagesSent.Load(),
"messages_received": m.messagesReceived.Load(),
"handoffs_initiated": m.handoffsInitiated.Load(),
"handoffs_accepted": m.handoffsAccepted.Load(),
"handoffs_rejected": m.handoffsRejected.Load(),
"handoffs_failed": m.handoffsFailed.Load(),
"elections_won": m.electionsWon.Load(),
"messages_sent": m.messagesSent.Load(),
"messages_received": m.messagesReceived.Load(),
"handoffs_initiated": m.handoffsInitiated.Load(),
"handoffs_accepted": m.handoffsAccepted.Load(),
"handoffs_rejected": m.handoffsRejected.Load(),
"handoffs_failed": m.handoffsFailed.Load(),
"elections_won": m.electionsWon.Load(),
// Gauges
"load_score": m.currentLoadScore.Load(),
"active_sessions": m.activeSessions.Load(),
"member_count": m.memberCount.Load(),
"load_score": m.currentLoadScore.Load(),
"active_sessions": m.activeSessions.Load(),
"member_count": m.memberCount.Load(),
// System info
"uptime_seconds": time.Since(m.startTime).Seconds(),
"uptime_seconds": time.Since(m.startTime).Seconds(),
// Latency histograms
"latency_ms": latency,
"latency_ms": latency,
}
}
@ -205,7 +205,7 @@ func (m *MetricsCollector) percentile(bucket *LatencyBucket, p float64) float64
cumulative := int64(0)
// Upper bounds for each bucket in ms
upperBounds := []int64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 1<<62}
upperBounds := []int64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 1 << 62}
for i, count := range bucket.buckets {
cumulative += count

40
pkg/swarm/node.go
View file

@ -63,13 +63,13 @@ const (
// NodeState represents the state of a node in the membership view.
type NodeState struct {
Node *NodeInfo `json:"node"`
Status NodeStatus `json:"status"`
StatusSince int64 `json:"status_since"` // Unix nano when status was set
LastSeen int64 `json:"last_seen"` // Unix nano of last sighting
LastPing int64 `json:"last_ping"` // Unix nano of last successful ping
PingSuccess int `json:"ping_success"` // Consecutive successful pings
PingFailure int `json:"ping_failure"` // Consecutive failed pings
Node *NodeInfo `json:"node"`
Status NodeStatus `json:"status"`
StatusSince int64 `json:"status_since"` // Unix nano when status was set
LastSeen int64 `json:"last_seen"` // Unix nano of last sighting
LastPing int64 `json:"last_ping"` // Unix nano of last successful ping
PingSuccess int `json:"ping_success"` // Consecutive successful pings
PingFailure int `json:"ping_failure"` // Consecutive failed pings
}
// IsAvailable returns true if the node is available for handoff.
@ -85,9 +85,9 @@ func (ns *NodeState) UpdateStatus(status NodeStatus) {
// NodeEvent represents a node state change event.
type NodeEvent struct {
Node *NodeInfo `json:"node"`
Event EventType `json:"event"`
Time int64 `json:"time"`
Node *NodeInfo `json:"node"`
Event EventType `json:"event"`
Time int64 `json:"time"`
}
// EventType represents the type of node event.
@ -195,13 +195,13 @@ func (ed *EventDispatcher) DispatchContext(event *NodeEvent, ctx context.Context
// NodeStats tracks statistics about a node.
type NodeStats struct {
MessagesSent int64 `json:"messages_sent"`
MessagesReceived int64 `json:"messages_received"`
HandoffsAccepted int `json:"handoffs_accepted"`
HandoffsInitiated int `json:"handoffs_initiated"`
LastError string `json:"last_error,omitempty"`
LastErrorTime time.Time `json:"last_error_time,omitempty"`
UptimeStart time.Time `json:"uptime_start"`
MessagesSent int64 `json:"messages_sent"`
MessagesReceived int64 `json:"messages_received"`
HandoffsAccepted int `json:"handoffs_accepted"`
HandoffsInitiated int `json:"handoffs_initiated"`
LastError string `json:"last_error,omitempty"`
LastErrorTime time.Time `json:"last_error_time,omitempty"`
UptimeStart time.Time `json:"uptime_start"`
}
// NodeWithState combines a node with its state and stats.
@ -222,9 +222,9 @@ func (nws *NodeWithState) IsAvailable() bool {
// ClusterView represents the current view of the cluster.
type ClusterView struct {
Nodes map[string]*NodeWithState `json:"nodes"`
LocalNodeID string `json:"local_node_id"`
Size int `json:"size"`
Version int64 `json:"version"` // View version for conflict detection
LocalNodeID string `json:"local_node_id"`
Size int `json:"size"`
Version int64 `json:"version"` // View version for conflict detection
mu sync.RWMutex
}

30
pkg/swarm/session_transfer.go
View file

@ -17,12 +17,12 @@ import (
// SessionTransfer handles session migration between nodes.
type SessionTransfer struct {
config RPCConfig
localNode *NodeInfo
transfers map[string]*TransferOperation // session_key -> operation
mu sync.RWMutex
conn *net.UDPConn
onReceive func(*TransferPayload)
config RPCConfig
localNode *NodeInfo
transfers map[string]*TransferOperation // session_key -> operation
mu sync.RWMutex
conn *net.UDPConn
onReceive func(*TransferPayload)
}
// TransferOperation represents an ongoing transfer operation.
@ -49,15 +49,15 @@ const (
// TransferPayload represents the session data being transferred.
type TransferPayload struct {
SessionKey string `json:"session_key"`
SourceNodeID string `json:"source_node_id"`
TargetNodeID string `json:"target_node_id"`
Messages []SessionMessage `json:"messages"`
Summary string `json:"summary,omitempty"`
Context map[string]any `json:"context,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
Timestamp int64 `json:"timestamp"`
TransferID string `json:"transfer_id"`
SessionKey string `json:"session_key"`
SourceNodeID string `json:"source_node_id"`
TargetNodeID string `json:"target_node_id"`
Messages []SessionMessage `json:"messages"`
Summary string `json:"summary,omitempty"`
Context map[string]any `json:"context,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
Timestamp int64 `json:"timestamp"`
TransferID string `json:"transfer_id"`
}
// NewSessionTransfer creates a new session transfer handler.

10
pkg/swarm/swarm_test.go
View file

@ -85,11 +85,11 @@ func TestClusterView(t *testing.T) {
func TestLoadMonitor(t *testing.T) {
config := &LoadMonitorConfig{
Enabled: true,
Interval: Duration{time.Second},
SampleSize: 10,
CPUWeight: 0.3,
MemoryWeight: 0.3,
Enabled: true,
Interval: Duration{time.Second},
SampleSize: 10,
CPUWeight: 0.3,
MemoryWeight: 0.3,
SessionWeight: 0.4,
}

16
pkg/swarm/types.go
View file

@ -9,17 +9,17 @@ package swarm
// SessionMessage represents a message in a session.
// This is shared across handoff and session transfer.
type SessionMessage struct {
Role string `json:"role"`
Content string `json:"content"`
Timestamp int64 `json:"timestamp,omitempty"`
Role string `json:"role"`
Content string `json:"content"`
Timestamp int64 `json:"timestamp,omitempty"`
ToolCalls []ToolCallData `json:"tool_calls,omitempty"`
}
// ToolCallData represents tool call information in a message.
type ToolCallData struct {
ID string `json:"id"`
Name string `json:"name"`
Arguments map[string]any `json:"arguments"`
Result string `json:"result,omitempty"`
Extra map[string]any `json:"extra,omitempty"`
ID string `json:"id"`
Name string `json:"name"`
Arguments map[string]any `json:"arguments"`
Result string `json:"result,omitempty"`
Extra map[string]any `json:"extra,omitempty"`
}