Files
BountyBoard/internal/extsvc/client.go
T
etalon f87b954f27 feat: atomization pipeline with job queue, atomizer client, WS hub, and board UI (phase 7)
- internal/extsvc: circuit breaker (§11.16) + retrying bearer JSON client
- atomizer client: §5.1 contract incl. defensive coefficient normalization
  and extension coefficient range (0,2]
- persisted jobs collection: atomic claim, panic recovery, exponential
  backoff (max 3), stale-running requeue, shutdown-safe bookkeeping
- subdivide (async 202, atomizing status, re-run replaces unpublished
  children after confirm) and extend (sibling task, source budget)
- failure path reverts status and notifies the consultant on final attempt
- PATCH task editing with bounty recompute, budget cascade to descendants
- publish single + bulk; task detail endpoint role-scoped
- coder/websocket hub: multiplexed channels, origin check, 30s heartbeats;
  client ws.js with reconnect + 15s polling fallback
- consultant atomization board: tree by root, coefficient sliders with live
  per-parent sum indicator, bounty preview, modals, shimmer, atomizer-down
  gating via /api/v1/service-health
- fix: tasks external-ref unique index is partial, not sparse (compound
  sparse indexed every task through customerId and broke child inserts)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-06-12 19:17:09 +02:00

151 lines
3.8 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package extsvc
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"math/rand/v2"
"net/http"
"time"
)
// ErrBreakerOpen is returned without touching the network when the circuit
// is open.
var ErrBreakerOpen = errors.New("circuit breaker open")
// ServiceError is a non-2xx response from the external service, carrying
// its §5 error envelope when present.
type ServiceError struct {
Status int
Code string
Message string
}
func (e *ServiceError) Error() string {
return fmt.Sprintf("service responded %d: %s %s", e.Status, e.Code, e.Message)
}
// Client is a bearer-token JSON client with retry (×2 on 5xx/network with
// jitter, §12) and a circuit breaker, shared by the two service clients.
type Client struct {
Name string
BaseURL func() string // resolved per call (admin override support)
Token string
HTTP *http.Client
Breaker *Breaker
}
func NewClient(name string, baseURL func() string, token string, timeout time.Duration) *Client {
return &Client{
Name: name,
BaseURL: baseURL,
Token: token,
HTTP: &http.Client{Timeout: timeout},
Breaker: NewBreaker(),
}
}
func (c *Client) BreakerState() string { return c.Breaker.State() }
// Healthy probes /healthz without breaker accounting (used by readyz and the
// admin panel).
func (c *Client) Healthy(ctx context.Context) error {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.BaseURL()+"/healthz", nil)
if err != nil {
return err
}
resp, err := c.HTTP.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("%s healthz: status %d", c.Name, resp.StatusCode)
}
return nil
}
// Call POSTs/GETs JSON through the breaker with retries. A nil out skips
// decoding.
func (c *Client) Call(ctx context.Context, method, path string, in, out any) error {
if !c.Breaker.Allow() {
return fmt.Errorf("%s: %w", c.Name, ErrBreakerOpen)
}
err := c.callOnce(ctx, method, path, in, out)
c.Breaker.Record(err)
return err
}
func (c *Client) callOnce(ctx context.Context, method, path string, in, out any) error {
var body []byte
if in != nil {
var err error
if body, err = json.Marshal(in); err != nil {
return fmt.Errorf("encode request: %w", err)
}
}
var lastErr error
for attempt := 0; attempt < 3; attempt++ {
if attempt > 0 {
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(time.Duration(attempt)*time.Second + time.Duration(rand.IntN(500))*time.Millisecond):
}
}
req, err := http.NewRequestWithContext(ctx, method, c.BaseURL()+path, bytes.NewReader(body))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+c.Token)
resp, err := c.HTTP.Do(req)
if err != nil {
lastErr = fmt.Errorf("%s %s: %w", c.Name, path, err)
continue // network error → retry
}
data, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20))
resp.Body.Close()
if err != nil {
lastErr = fmt.Errorf("%s %s: read body: %w", c.Name, path, err)
continue
}
if resp.StatusCode >= 500 {
lastErr = &ServiceError{Status: resp.StatusCode, Message: string(truncate(data, 200))}
continue // 5xx → retry
}
if resp.StatusCode >= 400 {
se := &ServiceError{Status: resp.StatusCode}
var env struct {
Error struct {
Code string `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
if json.Unmarshal(data, &env) == nil {
se.Code, se.Message = env.Error.Code, env.Error.Message
}
return se // 4xx is not retryable
}
if out == nil {
return nil
}
if err := json.Unmarshal(data, out); err != nil {
return fmt.Errorf("%s %s: decode response: %w", c.Name, path, err)
}
return nil
}
return lastErr
}
func truncate(b []byte, n int) []byte {
if len(b) <= n {
return b
}
return b[:n]
}