Compare commits
2 Commits
26ff646ace
...
bf2587c7fa
| Author | SHA1 | Date | |
|---|---|---|---|
| bf2587c7fa | |||
| d7a43afd90 |
@@ -15,6 +15,7 @@ import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
@@ -329,7 +330,15 @@ func Load(path string) (*Config, error) {
|
||||
// Expand ${VAR} or $VAR patterns from environment variables. This lets
|
||||
// secrets live in env (docker-compose env_file) rather than the config
|
||||
// file committed to git.
|
||||
expanded := os.ExpandEnv(string(b))
|
||||
expanded, missing := expandEnv(string(b))
|
||||
if len(missing) > 0 {
|
||||
// An unset variable expands to "", which every block reads as "not
|
||||
// configured" and none of them complains about. That is the intended
|
||||
// behaviour and it stays: CI parses this same file with no secrets
|
||||
// present. What was missing is the line telling the operator which
|
||||
// capability he just turned off by forgetting an env file.
|
||||
log.Printf("config: %s references unset environment variables %v — those settings are empty, so whatever they configure is off", path, missing)
|
||||
}
|
||||
var c Config
|
||||
if err := json.Unmarshal([]byte(expanded), &c); err != nil {
|
||||
return nil, fmt.Errorf("config: parse %s: %w", path, err)
|
||||
@@ -341,6 +350,23 @@ func Load(path string) (*Config, error) {
|
||||
return &c, nil
|
||||
}
|
||||
|
||||
// expandEnv is os.ExpandEnv plus the names it could not resolve, each reported
|
||||
// once and in the order the file mentions them. A variable set to the empty
|
||||
// string counts as set: the operator wrote it down, so he meant it.
|
||||
func expandEnv(s string) (string, []string) {
|
||||
var missing []string
|
||||
seen := map[string]bool{}
|
||||
out := os.Expand(s, func(name string) string {
|
||||
v, ok := os.LookupEnv(name)
|
||||
if !ok && !seen[name] {
|
||||
seen[name] = true
|
||||
missing = append(missing, name)
|
||||
}
|
||||
return v
|
||||
})
|
||||
return out, missing
|
||||
}
|
||||
|
||||
func (c *Config) applyDefaults() {
|
||||
if c.IntakeJournal == 0 {
|
||||
c.IntakeJournal = DefaultIntakeJournal
|
||||
|
||||
+33
-7
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"log"
|
||||
"net/http"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
@@ -46,6 +47,7 @@ type Pair struct {
|
||||
interval time.Duration
|
||||
http *http.Client
|
||||
stop chan struct{}
|
||||
stopOnce sync.Once
|
||||
}
|
||||
|
||||
// ErrRemoteUnavailable — the workstation model was required and is not
|
||||
@@ -107,13 +109,11 @@ func (p *Pair) Start(ctx context.Context) {
|
||||
}()
|
||||
}
|
||||
|
||||
// Stop ends the prober. Idempotent.
|
||||
// Stop ends the prober. Idempotent, and safe from two goroutines at once. The
|
||||
// check-then-close it replaced let both callers see an open channel and the
|
||||
// second close panicked, which turned a shutdown race into a crash.
|
||||
func (p *Pair) Stop() {
|
||||
select {
|
||||
case <-p.stop:
|
||||
default:
|
||||
close(p.stop)
|
||||
}
|
||||
p.stopOnce.Do(func() { close(p.stop) })
|
||||
}
|
||||
|
||||
// Available reports whether the workstation will take work right now. It reads
|
||||
@@ -170,7 +170,9 @@ func (p *Pair) Complete(ctx context.Context, r Req) (string, error) {
|
||||
}
|
||||
why := "workstation down"
|
||||
if p.Available() {
|
||||
out, err := p.remote.Complete(ctx, r)
|
||||
rctx, cancel := remoteBudget(ctx)
|
||||
out, err := p.remote.Complete(rctx, r)
|
||||
cancel()
|
||||
if err == nil {
|
||||
log.Print("llm: served by the workstation model")
|
||||
return out, nil
|
||||
@@ -184,6 +186,30 @@ func (p *Pair) Complete(ctx context.Context, r Req) (string, error) {
|
||||
return p.floor.Complete(ctx, r)
|
||||
}
|
||||
|
||||
// remoteBudget bounds the workstation attempt so the floor still has time to
|
||||
// answer. A turn carrying a deadline used to hand the whole of it to the
|
||||
// remote, so a workstation that accepted the connection and then hung ate the
|
||||
// budget and the fallback ran on an already-expired context: the floor
|
||||
// returned the deadline error and the turn broke on the workstation being
|
||||
// slow, which docs/offload.md says must never happen. Half is the split
|
||||
// because both halves have to be able to finish, and there is no reason to
|
||||
// prefer either one when the remote is the part that failed.
|
||||
//
|
||||
// A context with no deadline is left alone. The remote client's own timeout
|
||||
// (workstation.timeout, 90s by default) bounds it there, and shortening that
|
||||
// silently would change the configured budget.
|
||||
func remoteBudget(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||
dl, ok := ctx.Deadline()
|
||||
if !ok {
|
||||
return ctx, func() {}
|
||||
}
|
||||
left := time.Until(dl)
|
||||
if left <= 0 {
|
||||
return ctx, func() {}
|
||||
}
|
||||
return context.WithTimeout(ctx, left/2)
|
||||
}
|
||||
|
||||
// CompleteRemote runs r on the workstation or refuses. It never falls back,
|
||||
// because for a world question the resident 1.7B does not answer worse, it
|
||||
// invents. Callers turn ErrRemoteUnavailable into a named gap.
|
||||
|
||||
@@ -154,6 +154,48 @@ func TestRemoteErrorMidRequestFallsBack(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A workstation that accepts the connection and then hangs must not spend the
|
||||
// whole turn budget. It used to: the remote got the caller's context unchanged,
|
||||
// so the fallback ran on an expired one and the floor returned the deadline
|
||||
// error instead of an answer. The turn broke on the workstation being slow,
|
||||
// which is the one outcome docs/offload.md rules out.
|
||||
func TestHangingRemoteLeavesTheFloorABudget(t *testing.T) {
|
||||
var floorHits atomic.Int64
|
||||
// released, not r.Context().Done(): httptest.Server.Close waits for the
|
||||
// handler, and a handler that only watches the request context can outlive
|
||||
// the test when the client hangs up without the server noticing.
|
||||
released := make(chan struct{})
|
||||
hang := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
select {
|
||||
case <-released:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
}))
|
||||
defer hang.Close()
|
||||
defer close(released)
|
||||
floor := completionServer(t, "floor", &floorHits)
|
||||
up := &atomic.Bool{}
|
||||
up.Store(true)
|
||||
health := healthServer(t, up)
|
||||
|
||||
p := NewPair(New(hang.URL, time.Minute), New(floor.URL, time.Minute), health.URL, time.Hour)
|
||||
p.Start(context.Background())
|
||||
defer p.Stop()
|
||||
if !waitFor(t, p.Available) {
|
||||
t.Fatal("prober never saw the remote come up")
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 400*time.Millisecond)
|
||||
defer cancel()
|
||||
out, err := p.Complete(ctx, Req{User: "привет"})
|
||||
if err != nil {
|
||||
t.Fatalf("complete: %v", err)
|
||||
}
|
||||
if out != "floor" || floorHits.Load() != 1 {
|
||||
t.Fatalf("out = %q, floor hits = %d", out, floorHits.Load())
|
||||
}
|
||||
}
|
||||
|
||||
// The naming half of the degradation rule. A world question must not be handed
|
||||
// to the resident model, because it answers by inventing.
|
||||
func TestCompleteRemoteNamesTheGap(t *testing.T) {
|
||||
|
||||
+18
-14
@@ -37,8 +37,9 @@ type Server struct {
|
||||
addr netaddr.Addr
|
||||
ln net.Listener
|
||||
|
||||
wg sync.WaitGroup
|
||||
done chan struct{}
|
||||
wg sync.WaitGroup
|
||||
done chan struct{}
|
||||
closeOnce sync.Once
|
||||
|
||||
// connCount — assigned per accepted conn, used in logs to distinguish
|
||||
// concurrent connections. Monotonic; not load-bearing for correctness.
|
||||
@@ -180,20 +181,23 @@ func (srv *Server) dispatch(ctx context.Context, req Request) (json.RawMessage,
|
||||
}
|
||||
|
||||
// Close stops accepting and waits for in-flight connections to drain. The
|
||||
// socket file is removed so a restart can rebind cleanly. Idempotent.
|
||||
// socket file is removed so a restart can rebind cleanly.
|
||||
//
|
||||
// Idempotent, and safe from two goroutines at once. The check-then-close it
|
||||
// replaced let both callers see an open channel and the second close panicked,
|
||||
// so a shutdown racing a signal handler took the process down the one way a
|
||||
// clean shutdown is supposed to prevent.
|
||||
func (srv *Server) Close() error {
|
||||
select {
|
||||
case <-srv.done:
|
||||
return nil
|
||||
default:
|
||||
var err error
|
||||
srv.closeOnce.Do(func() {
|
||||
close(srv.done)
|
||||
}
|
||||
if srv.ln == nil {
|
||||
return nil
|
||||
}
|
||||
err := srv.ln.Close()
|
||||
srv.wg.Wait()
|
||||
netaddr.Cleanup(srv.addr)
|
||||
if srv.ln == nil {
|
||||
return
|
||||
}
|
||||
err = srv.ln.Close()
|
||||
srv.wg.Wait()
|
||||
netaddr.Cleanup(srv.addr)
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user