client-exec CLI

internal/clients/clients.go

@@ -23,8 +23,11 @@ const (
 	pingInterval = 30 * time.Second
 	pingTimeout  = 15 * time.Second
 	expireAfter  = 24 * time.Hour
+	ackTimeout   = 2 * time.Second
 )
 
+var ErrUnavailable = errors.New("client unavailable")
+
 type Client struct {
 	ID           string
 	Login        string
@@ -150,9 +153,12 @@ func (r *Registry) Serve(ctx context.Context, ws *websocket.Conn, id, login, hos
 			ws.Close(websocket.StatusUnsupportedData, "messages must be ClientReply")
 			return
 		}
+		accepted := reply.Accepted != nil && *reply.Accepted
 		c.mu.Lock()
 		ch := c.pending[reply.Id]
-		delete(c.pending, reply.Id)
+		if !accepted {
+			delete(c.pending, reply.Id)
+		}
 		c.mu.Unlock()
 		if ch == nil {
 			slog.Warn("client reply for no request", "host", host, "id", reply.Id)
@@ -204,6 +210,15 @@ func (r *Registry) List() []Client {
 	return out
 }
 
+func (r *Registry) Recent() []Client {
+	rows := r.List()
+	sort.SliceStable(rows, func(i, j int) bool {
+		a, b := rows[i].LastActionAt, rows[j].LastActionAt
+		return a != nil && (b == nil || a.After(*b))
+	})
+	return rows
+}
+
 func (r *Registry) Call(ctx context.Context, name string, req api.ClientRequest) (api.ClientReply, error) {
 	r.mu.Lock()
 	matches := []entry{}
@@ -220,7 +235,7 @@ func (r *Registry) Call(ctx context.Context, name string, req api.ClientRequest)
 	r.mu.Unlock()
 	slices.Sort(connected)
 	if len(matches) == 0 {
-		return api.ClientReply{}, fmt.Errorf("no client named %q is connected; connected: %s", name, strings.Join(connected, ", "))
+		return api.ClientReply{}, fmt.Errorf("%w: no client named %q is connected; connected: %s", ErrUnavailable, name, strings.Join(connected, ", "))
 	}
 	if len(matches) > 1 {
 		matched := make([]string, len(matches))
@@ -233,7 +248,7 @@ func (r *Registry) Call(ctx context.Context, name string, req api.ClientRequest)
 	c := matches[0].conn
 
 	req.Id = ids.New(ids.ClientRequest)
-	ch := make(chan api.ClientReply, 1)
+	ch := make(chan api.ClientReply, 2)
 	c.mu.Lock()
 	c.pending[req.Id] = ch
 	c.mu.Unlock()
@@ -252,16 +267,26 @@ func (r *Registry) Call(ctx context.Context, name string, req api.ClientRequest)
 		return api.ClientReply{}, fmt.Errorf("send to %s: %w", name, err)
 	}
 
-	select {
-	case reply := <-ch:
-		if reply.Error != nil {
-			return reply, errors.New(*reply.Error)
+	ack := time.NewTimer(ackTimeout)
+	defer ack.Stop()
+	for {
+		select {
+		case reply := <-ch:
+			if reply.Accepted != nil && *reply.Accepted {
+				ack.Stop()
+				continue
+			}
+			if reply.Error != nil {
+				return reply, errors.New(*reply.Error)
+			}
+			return reply, nil
+		case <-ack.C:
+			return api.ClientReply{}, fmt.Errorf("%w: %s didn't acknowledge the request within %s", ErrUnavailable, name, ackTimeout)
+		case <-c.done:
+			return api.ClientReply{}, fmt.Errorf("%w: %s disconnected before replying", ErrUnavailable, name)
+		case <-ctx.Done():
+			return api.ClientReply{}, fmt.Errorf("%s didn't reply: %w", name, ctx.Err())
 		}
-		return reply, nil
-	case <-c.done:
-		return api.ClientReply{}, fmt.Errorf("%s disconnected before replying", name)
-	case <-ctx.Done():
-		return api.ClientReply{}, fmt.Errorf("%s didn't reply: %w", name, ctx.Err())
 	}
 }