Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -89,14 +89,18 @@ reasoning encrypted, so there only the token counts show up.

auth: bearer `sk-lanv-...` from `key add`. fresh data dir with zero keys
runs open (single-user localhost); the moment one key exists, all `/v1/*`
require it. one stable `x-opencode-session` is derived per api key:
minting a fresh session per request burns free quota into 429s.
require it. on a public endpoint set `requireKey` in the config (or
`LANVELLO_REQUIRE_KEY=1`) so an empty data dir refuses instead of serving
open. keys.json is re-read when it changes, so `key add` / `key revoke`
take effect without a restart. one stable `x-opencode-session` is derived
per api key: minting a fresh session per request burns free quota into 429s.

## limits

429 retires the lane exit until `retry-after` and rotates country, request is
retried on another lane. 502/503/504 retry on another lane. per-request
proof goes to `$dataDir/proof.jsonl`.
429 retires the lane exit until `retry-after`; with tor lanes the process is
respawned in the background for a fresh exit and the lane returns once it is
up, while the request is retried on another lane. 502/503/504 retry on
another lane. per-request proof goes to `$dataDir/proof.jsonl`.

a lane stays busy for the whole answer, not just while the request is in
flight, so N lanes means N concurrent answers; the rest wait in the queue up
Expand Down
6 changes: 6 additions & 0 deletions cmd/lanvello/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,11 @@ func cmdServe(args []string) {
// tor-only: direct egress is gone. without tor lanes there is no service.
if len(cfg.Socks) == 0 {
tor.Ensure(m, cfg.DataDir, 52001, 52301, 90*time.Second)
// A 429 rotates the lane: respawn its tor in the background so the
// next request lands on a fresh exit.
m.OnRotate = func(l *lanes.Lane) {
tor.Respawn(m, l, cfg.DataDir, 52001, 52301, 90*time.Second)
}
}
if n := healthySocks(m); n == 0 {
fmt.Fprintln(os.Stderr, "no tor lanes up (tor binary missing or bootstrap failed) -- refusing to serve direct")
Expand All @@ -132,6 +137,7 @@ func cmdServe(args []string) {
up.WantCountry[pref] = cc
}
srv := server.New(ks, m, up, cfg.DataDir)
srv.RequireKey = cfg.RequireKey
fmt.Printf("lanvello %s listening on http://%s lanes=%d tor-only, free, no login\n", version, cfg.Listen, cfg.Lanes)
if err := http.ListenAndServe(cfg.Listen, srv.Handler()); err != nil {
fmt.Fprintln(os.Stderr, err)
Expand Down
16 changes: 16 additions & 0 deletions internal/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"encoding/json"
"os"
"path/filepath"
"strings"
)

type Config struct {
Expand All @@ -21,6 +22,9 @@ type Config struct {
// CatalogBudgetS bounds the lane wait for /v1/models, which must stay
// snappy even when every lane is busy with a long answer.
CatalogBudgetS float64 `json:"catalogBudgetS"`
// RequireKey refuses keyless /v1 requests even before the first key is
// issued: a public deployment must never fall back to open mode.
RequireKey bool `json:"requireKey"`
}

func DefaultDataDir() string {
Expand All @@ -44,9 +48,18 @@ func Defaults() Config {
Countries: []string{"us", "de", "nl", "fr", "ro", "gb", "ca", "se", "pl", "ch"},
WaitBudgetS: 90,
CatalogBudgetS: 20,
RequireKey: envBool("LANVELLO_REQUIRE_KEY"),
}
}

func envBool(k string) bool {
switch strings.ToLower(strings.TrimSpace(os.Getenv(k))) {
case "1", "true", "yes", "on":
return true
}
return false
}

func envOr(k, d string) string {
if v := os.Getenv(k); v != "" {
return v
Expand Down Expand Up @@ -100,6 +113,9 @@ func LoadFile(path string, base Config) (Config, error) {
if f.CatalogBudgetS > 0 {
base.CatalogBudgetS = f.CatalogBudgetS
}
if f.RequireKey {
base.RequireKey = true
}
if len(f.ModelLanes) > 0 {
if base.ModelLanes == nil {
base.ModelLanes = map[string]string{}
Expand Down
14 changes: 14 additions & 0 deletions internal/config/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -73,3 +73,17 @@ func TestLoadFileRejectsGarbage(t *testing.T) {
t.Fatal("garbage config must be an error, not silently defaulted")
}
}

func TestLoadFileRequireKey(t *testing.T) {
p := filepath.Join(t.TempDir(), "c.json")
if err := os.WriteFile(p, []byte(`{"requireKey":true}`), 0o600); err != nil {
t.Fatal(err)
}
cfg, err := LoadFile(p, Defaults())
if err != nil {
t.Fatal(err)
}
if !cfg.RequireKey {
t.Fatalf("requireKey lost: %+v", cfg)
}
}
48 changes: 47 additions & 1 deletion internal/keys/keys.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,11 @@ type Store struct {
path string
mu sync.Mutex
en []Entry
// mtime/size of the keys.json the in-memory list came from. A change on
// disk (lanvello key add/revoke runs in another process) is picked up on
// the next request instead of requiring a restart.
mtime int64
size int64
}

func Path(dataDir string) string { return filepath.Join(dataDir, "keys.json") }
Expand All @@ -48,9 +53,42 @@ func Open(dataDir string) (*Store, error) {
if err := json.Unmarshal(b, &s.en); err != nil {
return nil, err
}
s.stamp()
return s, nil
}

// stamp records the current file identity so maybeReload can tell later
// writes apart. Best-effort: a missing file simply never matches.
func (s *Store) stamp() {
if st, err := os.Stat(s.path); err == nil {
s.mtime, s.size = st.ModTime().UnixNano(), st.Size()
}
}

// maybeReload re-reads keys.json when another process changed it. Callers
// hold s.mu. Best-effort: a torn or unparsable read keeps the in-memory list.
func (s *Store) maybeReload() {
st, err := os.Stat(s.path)
if err != nil {
return
}
if st.ModTime().UnixNano() == s.mtime && st.Size() == s.size {
return
}
b, err := os.ReadFile(s.path)
if err != nil {
return
}
var en []Entry
if len(b) > 0 {
if err := json.Unmarshal(b, &en); err != nil {
return
}
}
s.en = en
s.mtime, s.size = st.ModTime().UnixNano(), st.Size()
}

func (s *Store) save() error {
tmp := s.path + ".tmp"
b, err := json.MarshalIndent(s.en, "", " ")
Expand All @@ -60,7 +98,11 @@ func (s *Store) save() error {
if err := os.WriteFile(tmp, b, 0o600); err != nil {
return err
}
return os.Rename(tmp, s.path)
if err := os.Rename(tmp, s.path); err != nil {
return err
}
s.stamp()
return nil
}

func hashOf(secret string) string {
Expand All @@ -72,6 +114,7 @@ func hashOf(secret string) string {
func (s *Store) Add(name string) (string, Entry, error) {
s.mu.Lock()
defer s.mu.Unlock()
s.maybeReload()
var raw [24]byte
if _, err := rand.Read(raw[:]); err != nil {
return "", Entry{}, err
Expand Down Expand Up @@ -100,6 +143,7 @@ func (s *Store) Add(name string) (string, Entry, error) {
func (s *Store) List() []Entry {
s.mu.Lock()
defer s.mu.Unlock()
s.maybeReload()
cp := make([]Entry, len(s.en))
copy(cp, s.en)
return cp
Expand All @@ -108,6 +152,7 @@ func (s *Store) List() []Entry {
func (s *Store) Revoke(name string) bool {
s.mu.Lock()
defer s.mu.Unlock()
s.maybeReload()
found := false
out := s.en[:0]
for _, x := range s.en {
Expand All @@ -129,6 +174,7 @@ func (s *Store) Revoke(name string) bool {
func (s *Store) Verify(secret string, allowEmpty bool) bool {
s.mu.Lock()
defer s.mu.Unlock()
s.maybeReload()
if secret == "" {
return false
}
Expand Down
34 changes: 34 additions & 0 deletions internal/keys/keys_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,3 +29,37 @@ func TestAddVerifyRevoke(t *testing.T) {
t.Fatal("revoked still verifies")
}
}

// The running server must see keys added or revoked by the CLI (`lanvello
// key add`, a separate process) without a restart.
func TestStorePicksUpExternalChanges(t *testing.T) {
dir := t.TempDir()
srv, err := keys.Open(dir)
if err != nil {
t.Fatal(err)
}
cli, err := keys.Open(dir)
if err != nil {
t.Fatal(err)
}
sec, _, err := cli.Add("atractio")
if err != nil {
t.Fatal(err)
}
if !srv.Verify(sec, false) {
t.Fatal("a key added by another process must verify without a restart")
}
var names []string
for _, e := range srv.List() {
names = append(names, e.Name)
}
if len(names) != 1 || names[0] != "atractio" {
t.Fatalf("list=%v", names)
}
if !cli.Revoke("atractio") {
t.Fatal("revoke failed")
}
if srv.Verify(sec, false) {
t.Fatal("a key revoked by another process must stop verifying")
}
}
58 changes: 52 additions & 6 deletions internal/lanes/lanes.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ type Lane struct {
Score int
Healthy bool
Healing bool
Rotating bool
LastUsed time.Time
LastReal time.Time
cmd *exec.Cmd
Expand All @@ -52,6 +53,10 @@ type Manager struct {
fallback []string
noTor bool
seq int
// OnRotate, when set (tor-spawned lanes), respawns the lane's process so
// a 429 lands on a fresh exit. The lane stays retired until RotateDone
// reports the new exit.
OnRotate func(*Lane)
}

func New(dataDir string, countries, fallback []string, socks []string, noTor bool, count int) *Manager {
Expand Down Expand Up @@ -182,7 +187,7 @@ func (m *Manager) pick(exclude map[int]bool, cc string) *Lane {
if exclude != nil && exclude[l.Index] {
continue
}
if !l.Healthy || l.Healing {
if !l.Healthy || l.Healing || l.Rotating {
continue
}
if cc != "" && l.Country != cc {
Expand Down Expand Up @@ -265,10 +270,12 @@ func (m *Manager) NoteLimited(l *Lane, retryAfter time.Duration) time.Time {
return l.LimitedTill
}

// Rotate moves lane to the best country with spare capacity (simple score sort).
// Rotate retires the lane's exit after a 429 and asks OnRotate (tor respawn,
// when the lanes are tor-spawned) for a fresh one in the best-scoring
// country. The retry-after limit is kept until the new exit is up — clearing
// it here handed the same limited exit straight back to the next request.
func (m *Manager) Rotate(l *Lane) string {
m.mu.Lock()
defer m.mu.Unlock()
pool := append(append([]string{}, m.countries...), m.fallback...)
seen := map[string]bool{}
var ranked []string
Expand All @@ -291,12 +298,51 @@ func (m *Manager) Rotate(l *Lane) string {
continue
}
l.Country = c
l.LimitedTill = time.Time{}
return c
break
}
hook := m.OnRotate
if hook != nil && !l.Rotating {
l.Rotating = true
go hook(l)
}
cc := l.Country
m.mu.Unlock()
return cc
}

// RotateDone reports a lane's fresh exit (a respawn finished): the lane is
// usable again and its retry-after limit is lifted.
func (m *Manager) RotateDone(l *Lane, socks, ip string) {
m.mu.Lock()
defer m.mu.Unlock()
if socks != "" {
l.SocksAddr = socks
l.Healthy = true
}
return l.Country
l.ExitIP = ip
l.Rotating = false
l.LimitedTill = time.Time{}
}

// RotateFailed clears the rotating flag after a failed respawn. The lane
// keeps its retry-after limit (with a cooldown floor) so the next request
// does not immediately try to rotate it again.
func (m *Manager) RotateFailed(l *Lane) {
m.mu.Lock()
defer m.mu.Unlock()
l.Rotating = false
if floor := time.Now().Add(5 * time.Minute); l.LimitedTill.Before(floor) {
l.LimitedTill = floor
}
}

// SetProcess records the tor process backing this lane so rotation can
// replace it; nil for lanes that are not tor-spawned.
func (l *Lane) SetProcess(cmd *exec.Cmd) { l.cmd = cmd }

// Process returns the tor process backing this lane, if any.
func (l *Lane) Process() *exec.Cmd { return l.cmd }

func (m *Manager) Emit(p Proof) {
m.mu.Lock()
path := m.proofPath
Expand Down
Loading
Loading