ai/byre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Reconnect the app's command channel after sleep; list clients from it

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-18 09:16:31 -07:00
parent 931d1b9a96
commit 1363a3604d
11 changed files with 444 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/clients/clients.go
+97 -32
View File
@@ -22,13 +22,16 @@ const (
readLimit = 256 << 20
pingInterval = 30 * time.Second
pingTimeout = 15 * time.Second
expireAfter = 24 * time.Hour
)
type Client struct {
Login string
Host string
Commit string
ConnectedAt time.Time
ID string
Login string
Host string
Commit string
ConnectedAt time.Time
LastActionAt *time.Time
}
type Transfer struct {
@@ -38,44 +41,91 @@ type Transfer struct {
}
type conn struct {
Client
ws *websocket.Conn
done chan struct{}
mu sync.Mutex
pending map[string]chan api.ClientReply
}
type entry struct {
Client
conn *conn
seen time.Time
notified time.Time
}
type Registry struct {
mu sync.Mutex
conns map[string]*conn
clients map[string]*entry
transfers map[string]Transfer
changed func()
}
func New() *Registry {
return &Registry{conns: map[string]*conn{}, transfers: map[string]Transfer{}}
func New(changed func()) *Registry {
return &Registry{clients: map[string]*entry{}, transfers: map[string]Transfer{}, changed: changed}
}
func (r *Registry) Serve(ctx context.Context, ws *websocket.Conn, client Client) {
func (r *Registry) entry(id, login, host, commit string, now time.Time) *entry {
for k, e := range r.clients {
if e.conn == nil && now.Sub(e.seen) > expireAfter {
delete(r.clients, k)
}
}
e := r.clients[id]
if e == nil {
e = &entry{Client: Client{ID: id}}
r.clients[id] = e
}
e.Login = login
e.Host = host
e.Commit = commit
e.seen = now
return e
}
func (r *Registry) Acted(id, login, host, commit string) bool {
if id == "" {
return false
}
r.mu.Lock()
defer r.mu.Unlock()
now := time.Now()
e := r.entry(id, login, host, commit, now)
e.LastActionAt = &now
if now.Sub(e.notified) < time.Second {
return false
}
e.notified = now
return true
}
func (r *Registry) Serve(ctx context.Context, ws *websocket.Conn, id, login, host, commit string) {
ws.SetReadLimit(readLimit)
c := &conn{Client: client, ws: ws, done: make(chan struct{}), pending: map[string]chan api.ClientReply{}}
c := &conn{ws: ws, done: make(chan struct{}), pending: map[string]chan api.ClientReply{}}
r.mu.Lock()
old := r.conns[client.Host]
r.conns[client.Host] = c
now := time.Now().UTC()
e := r.entry(id, login, host, commit, now)
old := e.conn
e.conn = c
e.ConnectedAt = now
r.mu.Unlock()
if old != nil {
old.ws.Close(websocket.StatusGoingAway, "replaced by a newer connection from this host")
old.ws.Close(websocket.StatusGoingAway, "replaced by a newer connection from this client")
}
slog.Info("client connected", "host", client.Host, "login", client.Login, "commit", client.Commit)
slog.Info("client connected", "id", id, "host", host, "login", login, "commit", commit)
r.changed()
defer func() {
r.mu.Lock()
if r.conns[client.Host] == c {
delete(r.conns, client.Host)
if e.conn == c {
e.conn = nil
e.seen = time.Now()
}
r.mu.Unlock()
close(c.done)
slog.Info("client disconnected", "host", client.Host, "login", client.Login)
slog.Info("client disconnected", "id", id, "host", host, "login", login)
r.changed()
}()
ctx, cancel := context.WithCancel(ctx)
@@ -98,7 +148,7 @@ func (r *Registry) Serve(ctx context.Context, ws *websocket.Conn, client Client)
delete(c.pending, reply.Id)
c.mu.Unlock()
if ch == nil {
slog.Warn("client reply for no request", "host", client.Host, "id", reply.Id)
slog.Warn("client reply for no request", "host", host, "id", reply.Id)
continue
}
ch <- reply
@@ -128,25 +178,40 @@ func (r *Registry) List() []Client {
r.mu.Lock()
defer r.mu.Unlock()
out := []Client{}
for _, c := range r.conns {
out = append(out, c.Client)
for _, e := range r.clients {
if e.conn != nil {
out = append(out, e.Client)
}
}
sort.Slice(out, func(i, j int) bool { return out[i].Host < out[j].Host })
sort.Slice(out, func(i, j int) bool { return out[i].ConnectedAt.Before(out[j].ConnectedAt) })
return out
}
func (r *Registry) Call(ctx context.Context, host string, req api.ClientRequest) (api.ClientReply, error) {
func (r *Registry) Call(ctx context.Context, name string, req api.ClientRequest) (api.ClientReply, error) {
r.mu.Lock()
c := r.conns[host]
hosts := []string{}
for h := range r.conns {
hosts = append(hosts, h)
matches := []*conn{}
ids := []string{}
connected := []string{}
for _, e := range r.clients {
if e.conn == nil {
continue
}
connected = append(connected, fmt.Sprintf("%s (%s)", e.Host, e.ID))
if e.ID == name || e.Host == name {
matches = append(matches, e.conn)
ids = append(ids, e.ID)
}
}
r.mu.Unlock()
if c == nil {
slices.Sort(hosts)
return api.ClientReply{}, fmt.Errorf("no client named %q is connected; connected: %s", host, strings.Join(hosts, ", "))
slices.Sort(connected)
if len(matches) == 0 {
return api.ClientReply{}, fmt.Errorf("no client named %q is connected; connected: %s", name, strings.Join(connected, ", "))
}
if len(matches) > 1 {
slices.Sort(ids)
return api.ClientReply{}, fmt.Errorf("%d clients on %s are connected; name one by its id: %s", len(matches), name, strings.Join(ids, ", "))
}
c := matches[0]
req.Id = rand.Text()
ch := make(chan api.ClientReply, 1)
@@ -165,7 +230,7 @@ func (r *Registry) Call(ctx context.Context, host string, req api.ClientRequest)
}
err = c.ws.Write(ctx, websocket.MessageText, data)
if err != nil {
return api.ClientReply{}, fmt.Errorf("send to %s: %w", host, err)
return api.ClientReply{}, fmt.Errorf("send to %s: %w", name, err)
}
select {
@@ -175,9 +240,9 @@ func (r *Registry) Call(ctx context.Context, host string, req api.ClientRequest)
}
return reply, nil
case <-c.done:
return api.ClientReply{}, fmt.Errorf("%s disconnected before replying", host)
return api.ClientReply{}, fmt.Errorf("%s disconnected before replying", name)
case <-ctx.Done():
return api.ClientReply{}, fmt.Errorf("%s didn't reply: %w", host, ctx.Err())
return api.ClientReply{}, fmt.Errorf("%s didn't reply: %w", name, ctx.Err())
}
}