mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-09 18:57:21 +00:00
Compare commits
66 Commits
archive/pr
...
v1.9.5
| Author | SHA1 | Date | |
|---|---|---|---|
| 2cf7808e63 | |||
| aaad52acc1 | |||
| 06f8c5dba4 | |||
| 4a0a1f96e4 | |||
| ac9746ec83 | |||
| 2272b4ecbd | |||
| ff8afaa43d | |||
| e5af4ec34c | |||
| 360a2a8644 | |||
| 2fd536e99b | |||
| 700f4179fa | |||
| 91e3013cb5 | |||
| 92b81ecfd3 | |||
| d119d09929 | |||
| 27b908d98c | |||
| ba10f03fc7 | |||
| cab3361073 | |||
| 35b9422232 | |||
| 2b34c7dc8a | |||
| 85c81c63cc | |||
| a2bb6de9fb | |||
| 6e63e4761d | |||
| 37545b6dbb | |||
| 314f4da8f2 | |||
| b93799a919 | |||
| c36200b78b | |||
| ba6de89d57 | |||
| 3b28fde22e | |||
| 9823773330 | |||
| 5eef5ad635 | |||
| 40636abcc2 | |||
| 22c9215b02 | |||
| e76e1623e4 | |||
| 0bf0653d00 | |||
| 6448934096 | |||
| 48e7e6b2a5 | |||
| be30ba01b3 | |||
| a21d0ca5a1 | |||
| 82d4e30b9f | |||
| cf14f66ad4 | |||
| c5079747b7 | |||
| 4e3b905b58 | |||
| 130b3f2e8d | |||
| dac0b4ce93 | |||
| 5a3b5d357c | |||
| a9d1bdeafb | |||
| d547bed457 | |||
| 97864895d4 | |||
| 31090fec81 | |||
| f9cd0770a9 | |||
| 6311913812 | |||
| aa003e84d3 | |||
| 28149bdd1a | |||
| 0cd1cd51f6 | |||
| 4b07be67b4 | |||
| 913974bb6c | |||
| ffb2b8f2f2 | |||
| 1fe379f630 | |||
| b03b12148f | |||
| a1d4365515 | |||
| 3548745f98 | |||
| 3fe0df5bf2 | |||
| 8ca59b4bb2 | |||
| 980408d045 | |||
| 4b37e1a0db | |||
| ebe60028f5 |
32
ARCHIVE.md
32
ARCHIVE.md
@ -1,32 +0,0 @@
|
||||
# archive/pre-v1.9-quota-refactor
|
||||
|
||||
**这是一个归档分支,不要从它继续开发。**
|
||||
|
||||
它冻结的是 v1.9 之前的一段插件重构历史,保存于 2026-10-05。
|
||||
|
||||
## 为什么保留
|
||||
|
||||
分支上有 16 个提交没有以等价形式进入主线。逐个 patch-id 核对后,其中 **3 个
|
||||
是远端完全没有的**:
|
||||
|
||||
| commit | 内容 | 为什么没进主线 |
|
||||
|---|---|---|
|
||||
| `a6c6e6e` | 可见性、state 分区、注册式 HTTP 端点 | 后继版本 `35b9422` 是它的 rebase 等价版,但这个 hash 上还挂着另一些只存在于此分支的改动 |
|
||||
| `6bc3ff8` | ccquota 显示每月额度 | ccquota 后来被通用 `quota` 插件取代,这段额度 UI 没进主线 |
|
||||
| `b5f6cac` | ccquota 页面首次打开是空的 | 同上,是 ccquota 生命周期内的修复 |
|
||||
|
||||
其余 13 个已在主线以等价 patch 存在(`35b9422` / `cab3361` / `ba10f03` 等),
|
||||
它们保留在这里是为了让这段历史可追溯,不是为了再次使用。
|
||||
|
||||
## 现在的功能落点
|
||||
|
||||
| 当时在这里 | 现在在哪 |
|
||||
|---|---|
|
||||
| ccquota 多账户额度插件 | `internal/lua/plugins/quota.lua`(通用套餐余量插件) |
|
||||
| 可见性 / state 分区 / 按密钥授权 | `internal/lua/plugin_access.go`(`AccessRule`) |
|
||||
| 注册式 HTTP 端点 | `internal/lua/plugin_http_handler.go` + `AccessRule` |
|
||||
|
||||
## 当前发布线
|
||||
|
||||
- `main` — 下一个未发布版本
|
||||
- `release/v1.9.x` — v1.9.x 发布线(最新 tag `v1.9.5`)
|
||||
4
cmd/gui/package-lock.json
generated
4
cmd/gui/package-lock.json
generated
@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "modelrouter-gui",
|
||||
"version": "1.9.0",
|
||||
"version": "1.9.3",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "modelrouter-gui",
|
||||
"version": "1.9.0",
|
||||
"version": "1.9.3",
|
||||
"license": "Proprietary",
|
||||
"devDependencies": {
|
||||
"electron": "^33.0.0",
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "modelrouter-gui",
|
||||
"version": "1.9.0",
|
||||
"version": "1.9.5",
|
||||
"description": "ModelRouter Desktop — embedded ModelRouter core with tray",
|
||||
"author": "ModelRouter",
|
||||
"main": "main.js",
|
||||
|
||||
@ -195,6 +195,36 @@ func applyRule(prices map[string]interface{}, rule *config.BillingRule, srcName
|
||||
if t.CacheDiscount != nil {
|
||||
entry["cache_discount"] = *t.CacheDiscount
|
||||
}
|
||||
// reasoning is optional and stays ABSENT (not nil, not 0) when the
|
||||
// rule does not set it, which is what makes the billing plugin skip
|
||||
// the repricing branch entirely. Writing an explicit 0 here would
|
||||
// mean "thinking is free" and silently under-bill every request
|
||||
// from a provider that bills it like ordinary output.
|
||||
if strings.TrimSpace(t.Reasoning) != "" {
|
||||
rp, err := usdPerM(t.Reasoning)
|
||||
if err != nil {
|
||||
return fmt.Errorf("model %q reasoning: %w", modelID, err)
|
||||
}
|
||||
entry["reasoning_price"] = rp
|
||||
}
|
||||
if lc := t.LongContext; lc != nil {
|
||||
if lc.Above <= 0 {
|
||||
return fmt.Errorf("model %q long_context: above must be > 0", modelID)
|
||||
}
|
||||
above, err := longContextRate(lc, prompt)
|
||||
if err != nil {
|
||||
return fmt.Errorf("model %q long_context: %w", modelID, err)
|
||||
}
|
||||
entry["long_context"] = map[string]interface{}{
|
||||
"above": lc.Above,
|
||||
// Marginal, not blended. The plugin charges the tokens past the
|
||||
// threshold at this rate and the rest at the base rate, so the
|
||||
// threshold composes with a cache discount (which applies to the
|
||||
// cached portion) instead of forcing the plugin to guess how
|
||||
// the two interact.
|
||||
"prompt": above,
|
||||
}
|
||||
}
|
||||
if rule.Peak != nil {
|
||||
entry["peak"] = compilePeak(rule.Peak)
|
||||
}
|
||||
@ -262,3 +292,32 @@ func compilePeak(p *config.BillingPeak) map[string]interface{} {
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// longContextRate resolves the marginal per-token rate for the tokens ABOVE a
|
||||
// threshold, given the model's base per-million prompt price.
|
||||
//
|
||||
// Two ways to express it are supported because providers publish both: an
|
||||
// absolute rate (OpenAI quotes \$3.00/M past 200k) and a multiple (DeepSeek
|
||||
// quotes exactly 2x past 128k). Neither may be missing: a step with no rate is a
|
||||
// silent no-op, and an operator who wrote one meant something.
|
||||
func longContextRate(lc *config.BillingLongContext, basePerToken float64) (float64, error) {
|
||||
hasPrompt := strings.TrimSpace(lc.Prompt) != ""
|
||||
hasMult := lc.Multiplier > 0
|
||||
switch {
|
||||
case hasPrompt && hasMult:
|
||||
p, err := usdPerM(lc.Prompt)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
// Both given: the absolute rate wins, because it is the one the provider
|
||||
// actually quotes. Silently multiplying instead would produce a number
|
||||
// the operator never wrote.
|
||||
return p, nil
|
||||
case hasPrompt:
|
||||
return usdPerM(lc.Prompt)
|
||||
case hasMult:
|
||||
return basePerToken * lc.Multiplier, nil
|
||||
default:
|
||||
return 0, fmt.Errorf("needs either prompt or multiplier")
|
||||
}
|
||||
}
|
||||
|
||||
@ -86,3 +86,129 @@ func TestSpecificURLTokenRuleStillMarksItsSource(t *testing.T) {
|
||||
t.Error("an unrelated source was marked priced by a specific-url rule")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReasoningPriceIsOmittedUnlessSet pins the absence convention for the
|
||||
// thinking-token rate.
|
||||
//
|
||||
// A rule that says nothing about reasoning must produce a price entry with NO
|
||||
// reasoning_price key, so the billing plugin skips the repricing branch and
|
||||
// bills thinking at the ordinary output rate — which is what every provider
|
||||
// actually charges. Writing an explicit 0 instead would read as "thinking is
|
||||
// free" and silently under-bill every request from that model.
|
||||
func TestReasoningPriceIsOmittedUnlessSet(t *testing.T) {
|
||||
sources := []config.Source{{Name: "up", BaseURL: "https://up.example/v1"}}
|
||||
|
||||
// Case 1: no reasoning declared → the key must not appear.
|
||||
p1 := &config.BillingProfile{ID: "p", Rules: []config.BillingRule{{
|
||||
URL: "https://up.example/v1", Mode: "token",
|
||||
Models: map[string]config.BillingToken{
|
||||
"m": {Prompt: "1.00", Completion: "2.00"},
|
||||
},
|
||||
}}}
|
||||
prices, err := Compile(p1, sources)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
entry := prices["models"].(map[string]interface{})["m"].(map[string]interface{})
|
||||
if _, present := entry["reasoning_price"]; present {
|
||||
t.Errorf("reasoning_price present (%v) without a rule; the plugin would treat 0 as free",
|
||||
entry["reasoning_price"])
|
||||
}
|
||||
|
||||
// Case 2: declared → present, and converted to per-token like every other rate.
|
||||
p2 := &config.BillingProfile{ID: "p", Rules: []config.BillingRule{{
|
||||
URL: "https://up.example/v1", Mode: "token",
|
||||
Models: map[string]config.BillingToken{
|
||||
"m": {Prompt: "1.00", Completion: "2.00", Reasoning: "6.00"},
|
||||
},
|
||||
}}}
|
||||
prices2, err := Compile(p2, sources)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
e2 := prices2["models"].(map[string]interface{})["m"].(map[string]interface{})
|
||||
rp, ok := e2["reasoning_price"].(float64)
|
||||
if !ok {
|
||||
t.Fatalf("reasoning_price missing or wrong type: %#v", e2["reasoning_price"])
|
||||
}
|
||||
if rp != 6e-6 {
|
||||
t.Errorf("reasoning_price = %v, want 6e-6 (USD per TOKEN)", rp)
|
||||
}
|
||||
// A malformed rate must be rejected, not silently ignored: an operator who
|
||||
// typed "6 USD" expects either that value or an error, never a rule that
|
||||
// looks applied and bills nothing.
|
||||
p3 := &config.BillingProfile{ID: "p", Rules: []config.BillingRule{{
|
||||
URL: "https://up.example/v1", Mode: "token",
|
||||
Models: map[string]config.BillingToken{
|
||||
"m": {Prompt: "1.00", Completion: "2.00", Reasoning: "six dollars"},
|
||||
},
|
||||
}}}
|
||||
if _, err := Compile(p3, sources); err == nil {
|
||||
t.Error("a non-numeric reasoning rate must fail the compile")
|
||||
}
|
||||
}
|
||||
|
||||
// TestLongContextStepCompiles pins the long-context step's two accepted forms
|
||||
// and its rejection of a step that would silently do nothing.
|
||||
//
|
||||
// Providers publish both shapes — DeepSeek says "2x past 128k", OpenAI quotes
|
||||
// an absolute "$3.00/M past 200k" — so both must compile. Neither may be
|
||||
// missing: a step with no rate is a rule that LOOKS like it handles long
|
||||
// context and does not, which under-bills without any visible sign.
|
||||
func TestLongContextStepCompiles(t *testing.T) {
|
||||
sources := []config.Source{{Name: "up", BaseURL: "https://up.example/v1"}}
|
||||
|
||||
compile := func(lc config.BillingLongContext) (map[string]interface{}, error) {
|
||||
p := &config.BillingProfile{ID: "p", Rules: []config.BillingRule{{
|
||||
URL: "https://up.example/v1", Mode: "token",
|
||||
Models: map[string]config.BillingToken{
|
||||
"m": {Prompt: "1.00", Completion: "2.00", LongContext: &lc},
|
||||
},
|
||||
}}}
|
||||
prices, err := Compile(p, sources)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
e := prices["models"].(map[string]interface{})["m"].(map[string]interface{})
|
||||
return e["long_context"].(map[string]interface{}), nil
|
||||
}
|
||||
|
||||
// Absolute rate (OpenAI's published shape).
|
||||
lc1, err := compile(config.BillingLongContext{Above: 200000, Prompt: "3.00"})
|
||||
if err != nil {
|
||||
t.Fatalf("absolute long_context rate rejected: %v", err)
|
||||
}
|
||||
if lc1["above"] != 200000 || lc1["prompt"] != 3e-6 {
|
||||
t.Errorf("absolute step = %#v, want above=200000 prompt=3e-6", lc1)
|
||||
}
|
||||
|
||||
// Multiplier (DeepSeek's shape): 2x of the 1.00/M base.
|
||||
lc2, err := compile(config.BillingLongContext{Above: 128000, Multiplier: 2})
|
||||
if err != nil {
|
||||
t.Fatalf("multiplier long_context rejected: %v", err)
|
||||
}
|
||||
if lc2["prompt"] != 2e-6 {
|
||||
t.Errorf("multiplier step = %#v, want prompt=2e-6", lc2)
|
||||
}
|
||||
|
||||
// Both given: the absolute rate wins, because that is the number the
|
||||
// provider quotes. Multiplying instead would produce a figure the operator
|
||||
// never wrote.
|
||||
lc3, err := compile(config.BillingLongContext{Above: 128000, Prompt: "5.00", Multiplier: 2})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if lc3["prompt"] != 5e-6 {
|
||||
t.Errorf("both given = %#v, want the quoted absolute rate 5e-6", lc3)
|
||||
}
|
||||
|
||||
// Neither: a no-op step must fail, not compile into a rule that silently
|
||||
// bills long context at the base rate.
|
||||
if _, err := compile(config.BillingLongContext{Above: 128000}); err == nil {
|
||||
t.Error("a step with neither prompt nor multiplier must be rejected")
|
||||
}
|
||||
// A zero threshold would make every request "long".
|
||||
if _, err := compile(config.BillingLongContext{Above: 0, Multiplier: 2}); err == nil {
|
||||
t.Error("above=0 must be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
@ -453,6 +453,19 @@ type BillingToken struct {
|
||||
Prompt string `yaml:"prompt" json:"prompt"` // USD per million tokens, e.g. "0.15"
|
||||
Completion string `yaml:"completion" json:"completion"` // USD per million tokens
|
||||
CacheDiscount *float64 `yaml:"cache_discount,omitempty" json:"cache_discount,omitempty"`
|
||||
// Reasoning is the USD-per-million rate for thinking tokens, for the rare
|
||||
// provider that charges them differently from ordinary output. Omit it and
|
||||
// reasoning is billed at Completion — which is what OpenAI, DeepSeek and
|
||||
// Anthropic all do, and what every source on this deployment reports.
|
||||
//
|
||||
// It is a rate for a SUBSET of completion tokens, never an extra charge: the
|
||||
// billing plugin applies only the difference between this and Completion, so
|
||||
// setting it to a multiple of Completion reprices the thinking share rather
|
||||
// than billing it twice.
|
||||
Reasoning string `yaml:"reasoning,omitempty" json:"reasoning,omitempty"`
|
||||
// LongContext is the prompt-price step above a token threshold. Nil = one
|
||||
// flat rate, which is what a rule means when it says nothing.
|
||||
LongContext *BillingLongContext `yaml:"long_context,omitempty" json:"long_context,omitempty"`
|
||||
}
|
||||
|
||||
// BillingPeak declares peak-hour pricing.
|
||||
@ -464,6 +477,31 @@ type BillingPeak struct {
|
||||
Hours [][2]int `yaml:"hours,omitempty" json:"hours,omitempty"`
|
||||
}
|
||||
|
||||
// BillingLongContext is a prompt-price step for long inputs.
|
||||
//
|
||||
// Every major provider charges the input tokens ABOVE a threshold at a higher
|
||||
// rate — DeepSeek doubles past 128k, OpenAI and Anthropic step up as well. A
|
||||
// flat prompt price therefore under-bills exactly the traffic an agent workload
|
||||
// generates, and it under-bills it silently: the number is plausible, the bill
|
||||
// adds up, and nothing says the rate was wrong.
|
||||
//
|
||||
// The step applies to the PROMPT only. Output and cache reads are billed at
|
||||
// their own rates regardless of input length, which is how providers publish
|
||||
// it — the higher rate is for the extra context you made the model read, not for
|
||||
// what it wrote back.
|
||||
type BillingLongContext struct {
|
||||
// Above is the token count the step starts at. A request with exactly this
|
||||
// many prompt tokens is still at the base rate; the step covers the tokens
|
||||
// BEYOND it. Inclusive-below matches how the providers document it.
|
||||
Above int `yaml:"above" json:"above"`
|
||||
// Prompt is the USD-per-million rate for the tokens above Above.
|
||||
Prompt string `yaml:"prompt" json:"prompt"`
|
||||
// Multiplier is an alternative to Prompt: a factor applied to the base rate
|
||||
// (DeepSeek's published step is exactly 2x). When both are set Prompt wins,
|
||||
// and setting neither is a validation error rather than a silent no-op.
|
||||
Multiplier float64 `yaml:"multiplier,omitempty" json:"multiplier,omitempty"`
|
||||
}
|
||||
|
||||
// Validate rejects a BillingDSL that could not work as written.
|
||||
func (b *BillingDSL) Validate() error {
|
||||
if b == nil {
|
||||
|
||||
@ -789,6 +789,18 @@ func recordChatUsage(rec *Req, req *types.ChatRequest, resp *types.UnifiedRespon
|
||||
// Cache accounting: whichever source format the adapter normalized into
|
||||
// (prompt_tokens_details.cached_tokens or legacy prompt_cache_hit_tokens),
|
||||
// read it back so the request record carries the hit/miss split.
|
||||
// reasoning_tokens is a SUBSET of completion_tokens and is billed at the
|
||||
// completion rate by every provider measured, so it must NOT be added to
|
||||
// Compl. Recording it separately is what lets a price rule split thinking
|
||||
// from output for the providers that do price them differently, and what
|
||||
// keeps the billing plugin from guessing.
|
||||
//
|
||||
// Leaving it unrecorded is worse than useless: the adapters already parse it
|
||||
// and forward it (opencodego/gemini/anthropic all do), so the number exists
|
||||
// in the response and is thrown away at the one place that could keep it.
|
||||
if d := resp.TokenUsage.CompletionTokensDetails; d != nil && d.ReasoningTokens > 0 {
|
||||
rec.Reasoning = int64(d.ReasoningTokens)
|
||||
}
|
||||
if d := resp.TokenUsage.PromptTokensDetails; d != nil && d.CachedTokens > 0 {
|
||||
rec.CacheHit = int64(d.CachedTokens)
|
||||
}
|
||||
@ -1044,6 +1056,9 @@ func (g *Gateway) fireEnd(rec *Req) {
|
||||
// carried this (rec.CacheReported); the plugin payload did not, so any
|
||||
// plugin-level cache accounting had to guess.
|
||||
"cache_reported": rec.CacheReported,
|
||||
// A SUBSET of completion_tokens, not an extra. Billing plugins that
|
||||
// add it to completion would double-charge every thinking token.
|
||||
"reasoning_tokens": rec.Reasoning,
|
||||
"image_count": rec.ImageCount,
|
||||
"error": rec.Err,
|
||||
"time": rec.Time,
|
||||
|
||||
71
internal/gateway/chat_reasoning_test.go
Normal file
71
internal/gateway/chat_reasoning_test.go
Normal file
@ -0,0 +1,71 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"llmsproxy/internal/types"
|
||||
)
|
||||
|
||||
// TestRecordChatUsageKeepsReasoningTokens pins the reasoning-token plumbing.
|
||||
//
|
||||
// The adapters already parse completion_tokens_details.reasoning_tokens and
|
||||
// forward it (opencodego, gemini, anthropic all do), and types.go has carried
|
||||
// the field since it was defined — but nothing ever READ it. The number arrived
|
||||
// in the response and was discarded at the one place that could have kept it,
|
||||
// so a pricing rule had no way to split thinking from output.
|
||||
//
|
||||
// The property that matters most: reasoning is a SUBSET of completion. Every
|
||||
// provider measured bills it at the completion rate, so a billing plugin that
|
||||
// added it to completion_tokens would double-charge every thinking token. The
|
||||
// test asserts completion is left alone.
|
||||
func TestRecordChatUsageKeepsReasoningTokens(t *testing.T) {
|
||||
rec := &Req{}
|
||||
resp := &types.UnifiedResponse{
|
||||
TokenUsage: types.TokenUsage{
|
||||
Prompt: 100,
|
||||
Completion: 500,
|
||||
CompletionTokensDetails: &types.CompletionTokensDetails{
|
||||
ReasoningTokens: 320,
|
||||
},
|
||||
},
|
||||
}
|
||||
recordChatUsage(rec, &types.ChatRequest{}, resp)
|
||||
|
||||
if rec.Reasoning != 320 {
|
||||
t.Errorf("Reasoning = %d, want 320", rec.Reasoning)
|
||||
}
|
||||
// The subset must not be folded into completion — it is already counted there.
|
||||
if rec.Compl != 500 {
|
||||
t.Errorf("Compl = %d, want 500 (reasoning is a subset, never additive)", rec.Compl)
|
||||
}
|
||||
if rec.Prompt != 100 {
|
||||
t.Errorf("Prompt = %d, want 100", rec.Prompt)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRecordChatUsageReasoningAbsentStaysZero: "not reported" and "reported as
|
||||
// zero" must both leave the field at 0, and neither may invent a number. A
|
||||
// plugin reading Reasoning cannot tell them apart, which is why the field is
|
||||
// documented as "0 means not reported" rather than "0 means no thinking".
|
||||
func TestRecordChatUsageReasoningAbsentStaysZero(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
u types.TokenUsage
|
||||
}{
|
||||
{"no details object", types.TokenUsage{Prompt: 10, Completion: 20}},
|
||||
{"details without reasoning", types.TokenUsage{
|
||||
Prompt: 10,
|
||||
Completion: 20,
|
||||
CompletionTokensDetails: &types.CompletionTokensDetails{},
|
||||
}},
|
||||
} {
|
||||
rec := &Req{}
|
||||
recordChatUsage(rec, &types.ChatRequest{}, &types.UnifiedResponse{TokenUsage: tc.u})
|
||||
if rec.Reasoning != 0 {
|
||||
t.Errorf("%s: Reasoning = %d, want 0", tc.name, rec.Reasoning)
|
||||
}
|
||||
if rec.Compl != 20 {
|
||||
t.Errorf("%s: Compl = %d, want 20", tc.name, rec.Compl)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -117,6 +117,20 @@ func (g *Gateway) handlePluginsAPI(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// /api/plugins/{name}/fanout/{fn} — one call per argument set, in
|
||||
// parallel, from one request. See handlePluginFanout for why the
|
||||
// request-per-item form cannot be made concurrent from the client.
|
||||
if i := strings.Index(path, "/fanout/"); i >= 0 {
|
||||
name := path[:i]
|
||||
fn := path[i+len("/fanout/"):]
|
||||
if name == "" || fn == "" {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", "fanout needs a plugin and a function name")
|
||||
return
|
||||
}
|
||||
g.handlePluginFanout(w, r, name, fn)
|
||||
return
|
||||
}
|
||||
|
||||
if reqRole(r.Context()) != "admin" {
|
||||
writeError(w, http.StatusForbidden, "forbidden", "admin role required")
|
||||
return
|
||||
@ -133,7 +147,7 @@ func (g *Gateway) handlePluginsAPI(w http.ResponseWriter, r *http.Request) {
|
||||
// Visibility overrides and the endpoint access table, so the
|
||||
// plugin page renders what is IN FORCE rather than making the
|
||||
// operator read .lua files or config.yaml to find out.
|
||||
"visibility": ps.VisibilityOverrides(),
|
||||
"visibility": ps.VisibilityOverrides(),
|
||||
"http_access": ps.EndpointAccess(),
|
||||
})
|
||||
case http.MethodPost:
|
||||
@ -331,6 +345,68 @@ func (g *Gateway) handlePluginCall(w http.ResponseWriter, r *http.Request, name,
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{"result": result})
|
||||
}
|
||||
|
||||
// handlePluginFanout runs one plugin export per element of a list, in parallel,
|
||||
// from a SINGLE request.
|
||||
//
|
||||
// Why this exists: the quota page has twenty cards, and it used to fire twenty
|
||||
// browser requests at /call/. They arrived concurrently and then serialised
|
||||
// inside Plugins.Call, which holds the plugin mutex across the Lua function —
|
||||
// including its network wait. Measured on the live gateway: 20 probes took 29s of
|
||||
// wall clock while a single probe with no contention took 1.4–3.3s, so the page
|
||||
// sat on "probing…" for half a minute. The browser-side concurrency was real and
|
||||
// made no difference.
|
||||
//
|
||||
// A fan-out cannot be assembled from the client either: any request-per-item
|
||||
// arrangement is exactly the shape that queues. The parallelism has to happen
|
||||
// where the fan-out is received.
|
||||
func (g *Gateway) handlePluginFanout(w http.ResponseWriter, r *http.Request, name, fn string) {
|
||||
if reqRole(r.Context()) != "admin" {
|
||||
writeError(w, http.StatusForbidden, "forbidden", "admin role required")
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodPost {
|
||||
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "")
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
// Args is a LIST of argument objects, one per call. A bare object is
|
||||
// rejected rather than coerced: "args" meaning "one call" and "args"
|
||||
// meaning "many" depending on its shape is how a caller ends up probing
|
||||
// one source and believing it probed twenty.
|
||||
Args []map[string]interface{} `json:"args"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", "invalid json: "+err.Error())
|
||||
return
|
||||
}
|
||||
if len(req.Args) == 0 {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", "args must be a non-empty list of argument objects")
|
||||
return
|
||||
}
|
||||
// The page legitimately has one card per configured source. The bound is not
|
||||
// arbitrary: it stops one request from asking for thousands of VMs, and it is
|
||||
// enforced before any work starts rather than discovered shard by shard.
|
||||
if len(req.Args) > fanoutMaxArgs {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request",
|
||||
fmt.Sprintf("args has %d entries, limit is %d", len(req.Args), fanoutMaxArgs))
|
||||
return
|
||||
}
|
||||
results, err := g.core.Plugins().ShardCall(name, fn, req.Args)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadGateway, "plugin_error", err.Error())
|
||||
return
|
||||
}
|
||||
// A shard that failed contributes a null in its slot. That is deliberate and
|
||||
// is NOT the same as the whole request failing: one unreachable upstream must
|
||||
// not turn a page of twenty cards into a single error.
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{"results": results})
|
||||
}
|
||||
|
||||
// fanoutMaxArgs bounds one fan-out request. Twenty is the measured real source
|
||||
// count on the gateway this was built for; the headroom is for operators with
|
||||
// more sources configured.
|
||||
const fanoutMaxArgs = 64
|
||||
|
||||
func (g *Gateway) pluginDir() string {
|
||||
if c := g.core.Config(); c != nil {
|
||||
return c.PluginDir
|
||||
@ -369,7 +445,6 @@ func (g *Gateway) removePlugin(name string) error {
|
||||
return g.core.Plugins().Unload(name)
|
||||
}
|
||||
|
||||
|
||||
// handleEndpointAccess serves the operator's view and edit surface for one
|
||||
// plugin endpoint's reachable keys.
|
||||
//
|
||||
@ -439,7 +514,6 @@ func (g *Gateway) handleEndpointAccess(w http.ResponseWriter, r *http.Request, n
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// splitStateForKey rewrites a plugin state so that a non-admin caller sees only
|
||||
// the buckets that belong to them.
|
||||
//
|
||||
|
||||
@ -43,6 +43,16 @@ type Req struct {
|
||||
// accounting at all (even when the hit count is 0). The WebUI shows
|
||||
// "0%" instead of "—" for such rows.
|
||||
CacheReported bool `json:"cache_reported,omitempty"`
|
||||
// Reasoning is completion_tokens_details.reasoning_tokens.
|
||||
//
|
||||
// It is a SUBSET of Compl, not an addition: reasoning tokens are billed at
|
||||
// the completion rate and already counted there. This field exists so a
|
||||
// pricing rule can split them when a provider prices thinking tokens
|
||||
// differently (most do NOT — OpenAI and DeepSeek bill them as ordinary
|
||||
// output), and so a reader can see what fraction of a bill is thinking.
|
||||
// 0 means "not reported", which is NOT the same as "no reasoning happened" —
|
||||
// three of the sources measured return no such field at all.
|
||||
Reasoning int64 `json:"reasoning_tokens,omitempty"`
|
||||
OK bool `json:"ok"`
|
||||
// Status http status code
|
||||
Status int `json:"status"`
|
||||
|
||||
@ -5781,6 +5781,38 @@ toast(t("kAutoChainGlobal"));
|
||||
if (!r.ok) throw new Error(body.error?.message || r.status);
|
||||
return body.result;
|
||||
},
|
||||
// fanout runs one of the plugin's exported functions once per argument
|
||||
// set, IN PARALLEL, from a single request.
|
||||
//
|
||||
// `call` in a Promise.all is NOT this. Twenty concurrent /call/ requests
|
||||
// arrive concurrently and then serialise on the plugin mutex inside the
|
||||
// kernel — the lock is held across the Lua function, network wait
|
||||
// included, because one Lua VM has one stack. Measured on a live
|
||||
// gateway: twenty quota probes took 29s of wall clock through /call/,
|
||||
// against 1.4–3.3s for a single uncontended probe. The browser-side
|
||||
// concurrency bought nothing and the page looked simply broken.
|
||||
//
|
||||
// The parallelism has to happen where the fan-out is received, so it
|
||||
// takes one request. Results come back positionally aligned with `argSets`;
|
||||
// a failed item is null in its slot rather than an error for the batch,
|
||||
// because one unreachable upstream must not blank a whole page of cards.
|
||||
//
|
||||
// Same authorization as call: admin-only, and the function must be in
|
||||
// the plugin's exports.
|
||||
async fanout(name, fn, argSets) {
|
||||
const r = await fetch(
|
||||
"/api/plugins/" + encodeURIComponent(name) + "/fanout/" + encodeURIComponent(fn),
|
||||
{
|
||||
method: "POST",
|
||||
credentials: "same-origin",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ args: argSets || [] }),
|
||||
},
|
||||
);
|
||||
const body = await r.json().catch(() => ({}));
|
||||
if (!r.ok) throw new Error(body.error?.message || r.status);
|
||||
return body.results || [];
|
||||
},
|
||||
onTabShown(fn) {
|
||||
PLUGIN_TAB_CBS.__last = PLUGIN_TAB_CBS.__last || [];
|
||||
PLUGIN_TAB_CBS.__last.push(fn);
|
||||
|
||||
@ -994,3 +994,245 @@ func TestBillingCacheHitClampedInStats(t *testing.T) {
|
||||
tot["cache_fresh_tokens"])
|
||||
}
|
||||
}
|
||||
|
||||
// TestUnpricedIsTrackedPerDay pins the day-keyed unpriced tally.
|
||||
//
|
||||
// The lifetime counters cannot answer the question an operator actually has —
|
||||
// "is it STILL happening?" — because they keep the debt from before a rule was
|
||||
// added. After pricing a mispriced source the lifetime number never moves, so a
|
||||
// dashboard reading "unpriced: 6376" is indistinguishable from a system that is
|
||||
// still billing nothing. Day buckets are what make a pricing fix observable: the
|
||||
// offending day keeps its count and the next day starts at zero for it.
|
||||
func TestUnpricedIsTrackedPerDay(t *testing.T) {
|
||||
ps, _ := billingVM(t)
|
||||
// Nothing is priced in the default config, so every request is unpriced.
|
||||
// Two requests on two different UTC days.
|
||||
for _, ms := range []int64{1790000000000, 1790100000000} {
|
||||
ps.Fire(StageRequestEnd, map[string]interface{}{
|
||||
"model": "mystery-model", "source": "nowhere", "key": "***aaaaaa",
|
||||
"ok": true, "prompt_tokens": 10, "completion_tokens": 5,
|
||||
"time": ms,
|
||||
})
|
||||
}
|
||||
st := stateOf(t, ps)
|
||||
byDay, ok := st["unpriced_by_day"].(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("state has no unpriced_by_day map: %T", st["unpriced_by_day"])
|
||||
}
|
||||
if len(byDay) != 2 {
|
||||
t.Fatalf("expected 2 day buckets, got %d: %v", len(byDay), byDay)
|
||||
}
|
||||
var sum float64
|
||||
for day, bucket := range byDay {
|
||||
b := bucket.(map[string]interface{})
|
||||
sum += b["reqs"].(float64)
|
||||
models := b["models"].(map[string]interface{})
|
||||
if models["mystery-model"].(float64) != 1 {
|
||||
t.Errorf("%s: model tally = %v, want 1", day, models["mystery-model"])
|
||||
}
|
||||
}
|
||||
if sum != 2 {
|
||||
t.Errorf("day buckets sum to %v, want 2 (the lifetime counter must not be double-counted)", sum)
|
||||
}
|
||||
// The lifetime counters still exist and still agree — this is additive, not
|
||||
// a replacement.
|
||||
if st["unpriced_reqs"].(float64) != 2 {
|
||||
t.Errorf("unpriced_reqs = %v, want 2", st["unpriced_reqs"])
|
||||
}
|
||||
}
|
||||
|
||||
// TestReasoningBilledAtCompletionRateByDefault is the property that keeps the
|
||||
// standard formula matching a real invoice.
|
||||
//
|
||||
// reasoning_tokens is a subset of completion_tokens and every provider measured
|
||||
// (OpenAI, DeepSeek, Anthropic) charges it at the ordinary output rate. A rule
|
||||
// with no reasoning_price must therefore produce EXACTLY the number the
|
||||
// pre-reasoning formula produced — the field exists for a provider that prices
|
||||
// thinking differently, not as a new charge.
|
||||
func TestReasoningBilledAtCompletionRateByDefault(t *testing.T) {
|
||||
ps, _ := billingVM(t)
|
||||
if err := ps.SetState("billing", map[string]interface{}{
|
||||
"prices": map[string]interface{}{
|
||||
"currency": "USD",
|
||||
"models": map[string]interface{}{
|
||||
// USD PER TOKEN (compile.go divides the per-million figure by 1e6).
|
||||
// prompt 0.15/M, completion 0.60/M. No reasoning_price anywhere.
|
||||
"m": map[string]interface{}{"prompt": 0.15e-6, "completion": 0.60e-6},
|
||||
},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.Fire(StageRequestEnd, map[string]interface{}{
|
||||
"model": "m", "source": "s", "key": "***aaaaaa", "ok": true,
|
||||
"prompt_tokens": 1000, "completion_tokens": 1000,
|
||||
"reasoning_tokens": 800, // all of it thinking; still billed as output
|
||||
"time": 1750000000000,
|
||||
})
|
||||
st := stateOf(t, ps)
|
||||
cost := st["total"].(map[string]interface{})["cost"].(float64)
|
||||
approx(t, "reasoning at completion rate", cost, 0.00075)
|
||||
}
|
||||
|
||||
// TestReasoningPriceOverridesOnlyWhenSet: with a rule that DOES price thinking
|
||||
// differently, only the difference is applied. Charging the reasoning portion a
|
||||
// second time in full is the obvious wrong implementation and it doubles the
|
||||
// completion line.
|
||||
func TestReasoningPriceOverridesOnlyWhenSet(t *testing.T) {
|
||||
ps, _ := billingVM(t)
|
||||
if err := ps.SetState("billing", map[string]interface{}{
|
||||
"prices": map[string]interface{}{
|
||||
"currency": "USD",
|
||||
"models": map[string]interface{}{
|
||||
"m": map[string]interface{}{
|
||||
"prompt": 0.15e-6, "completion": 0.60e-6,
|
||||
// Thinking costs 3x ordinary output: 1.80/M.
|
||||
"reasoning_price": 1.80e-6,
|
||||
},
|
||||
},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.Fire(StageRequestEnd, map[string]interface{}{
|
||||
"model": "m", "source": "s", "key": "***aaaaaa", "ok": true,
|
||||
"prompt_tokens": 1000, "completion_tokens": 1000, "reasoning_tokens": 800,
|
||||
"time": 1750000000000,
|
||||
})
|
||||
st := stateOf(t, ps)
|
||||
cost := st["total"].(map[string]interface{})["cost"].(float64)
|
||||
// base = 1000*0.15e-6 + 1000*0.60e-6 = 0.00075
|
||||
// + 800 * (1.80-0.60)e-6 = 0.00096 → 0.00171
|
||||
approx(t, "reasoning repriced to 1.80/M", cost, 0.00171)
|
||||
}
|
||||
|
||||
// TestReasoningClampedToCompletion: a provider claiming MORE reasoning than
|
||||
// completion is malformed. Without a clamp the visible-output count goes
|
||||
// negative and the delta exceeds the whole request, so the bill silently
|
||||
// explodes on bad data instead of degrading.
|
||||
func TestReasoningClampedToCompletion(t *testing.T) {
|
||||
ps, _ := billingVM(t)
|
||||
if err := ps.SetState("billing", map[string]interface{}{
|
||||
"prices": map[string]interface{}{
|
||||
"currency": "USD",
|
||||
"models": map[string]interface{}{
|
||||
"m": map[string]interface{}{
|
||||
"prompt": 0.15e-6, "completion": 0.60e-6, "reasoning_price": 1.80e-6,
|
||||
},
|
||||
},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.Fire(StageRequestEnd, map[string]interface{}{
|
||||
"model": "m", "source": "s", "key": "***aaaaaa", "ok": true,
|
||||
"prompt_tokens": 1000, "completion_tokens": 1000, "reasoning_tokens": 99999,
|
||||
"time": 1750000000000,
|
||||
})
|
||||
st := stateOf(t, ps)
|
||||
cost := st["total"].(map[string]interface{})["cost"].(float64)
|
||||
// reasoning is clamped from 99999 down to completion (1000):
|
||||
// base = 1000*0.15e-6 + 1000*0.60e-6 = 0.00075
|
||||
// delta = 1000 * (1.80-0.60)e-6 = 0.00120
|
||||
// total = 0.00195
|
||||
// Without the clamp this would be 0.18 — 92x the whole request, and a
|
||||
// number plausible enough to survive a glance at a dashboard.
|
||||
approx(t, "reasoning clamped to completion", cost, 0.00195)
|
||||
}
|
||||
|
||||
// TestLongContextStepIsMarginal: the threshold reprices only the tokens PAST
|
||||
// it, at a marginal rate.
|
||||
//
|
||||
// A flat prompt price under-bills exactly the traffic an agent workload
|
||||
// produces — on this deployment 88% of requests exceed 128k with a median of
|
||||
// 247k — and it under-bills silently, because the number stays plausible. The
|
||||
// step must apply to the excess only; charging the WHOLE prompt at the higher
|
||||
// rate is the other obvious implementation and it overcharges short requests by
|
||||
// the same factor it was meant to fix long ones.
|
||||
func TestLongContextStepIsMarginal(t *testing.T) {
|
||||
|
||||
// Short request: below the threshold, so the step must not touch it at all.
|
||||
ps, _ := billingVM(t)
|
||||
if err := ps.SetState("billing", map[string]interface{}{
|
||||
"prices": map[string]interface{}{
|
||||
"currency": "USD",
|
||||
"models": map[string]interface{}{
|
||||
"m": map[string]interface{}{
|
||||
"prompt": 1e-6, "completion": 0,
|
||||
"long_context": map[string]interface{}{"above": 128000, "prompt": 2e-6},
|
||||
},
|
||||
},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.Fire(StageRequestEnd, map[string]interface{}{
|
||||
"model": "m", "source": "s", "key": "***aaaaaa", "ok": true,
|
||||
"prompt_tokens": 100000, "completion_tokens": 0, "time": 1750000000000,
|
||||
})
|
||||
got := stateOf(t, ps)["total"].(map[string]interface{})["cost"].(float64)
|
||||
approx(t, "short request unaffected", got, 0.1)
|
||||
|
||||
// Long request: 200k prompt with a 128k threshold → 72k tokens at the step.
|
||||
ps2, _ := billingVM(t)
|
||||
if err := ps2.SetState("billing", map[string]interface{}{
|
||||
"prices": map[string]interface{}{
|
||||
"currency": "USD",
|
||||
"models": map[string]interface{}{
|
||||
"m": map[string]interface{}{
|
||||
"prompt": 1e-6, "completion": 0,
|
||||
"long_context": map[string]interface{}{"above": 128000, "prompt": 2e-6},
|
||||
},
|
||||
},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps2.Fire(StageRequestEnd, map[string]interface{}{
|
||||
"model": "m", "source": "s", "key": "***aaaaaa", "ok": true,
|
||||
"prompt_tokens": 200000, "completion_tokens": 0, "time": 1750000000000,
|
||||
})
|
||||
got2 := stateOf(t, ps2)["total"].(map[string]interface{})["cost"].(float64)
|
||||
// 128000*1e-6 + 72000*2e-6 = 0.128 + 0.144 = 0.272
|
||||
approx(t, "long request marginal step", got2, 0.272)
|
||||
}
|
||||
|
||||
// TestLongContextComposesWithCacheDiscount pins the interaction, which is the
|
||||
// whole reason for computing the step as a marginal rate.
|
||||
//
|
||||
// On this deployment the cache hit rate is 88% and long prompts dominate, so
|
||||
// "cached tokens beyond the threshold" is the common case, not an edge case. A
|
||||
// cache read already bills at prompt*discount; the long-context rate applies on
|
||||
// top of THAT, not on top of the full prompt price. Computing it any other way
|
||||
// silently changes the price of the majority of traffic.
|
||||
func TestLongContextComposesWithCacheDiscount(t *testing.T) {
|
||||
ps, _ := billingVM(t)
|
||||
if err := ps.SetState("billing", map[string]interface{}{
|
||||
"prices": map[string]interface{}{
|
||||
"currency": "USD", "cache_discount": 0.1,
|
||||
"models": map[string]interface{}{
|
||||
"m": map[string]interface{}{
|
||||
"prompt": 1e-6, "completion": 0,
|
||||
"long_context": map[string]interface{}{"above": 100000, "prompt": 2e-6},
|
||||
},
|
||||
},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 200k prompt, 150k of it cached, 50k fresh. Excess past 100k = 100k, split
|
||||
// 25% fresh / 75% cached.
|
||||
ps.Fire(StageRequestEnd, map[string]interface{}{
|
||||
"model": "m", "source": "s", "key": "***aaaaaa", "ok": true,
|
||||
"prompt_tokens": 200000, "completion_tokens": 0,
|
||||
"cache_hit_tokens": 150000, "cache_reported": true,
|
||||
"time": 1750000000000,
|
||||
})
|
||||
got := stateOf(t, ps)["total"].(map[string]interface{})["cost"].(float64)
|
||||
// base: fresh 50000*1e-6 = 0.05
|
||||
// cache 150000*1e-6*0.1 = 0.015
|
||||
// step: fresh excess 25000*(2-1)e-6 = 0.025
|
||||
// cache excess 75000*(2-0.1)e-6 = 0.1425
|
||||
// total = 0.2325
|
||||
approx(t, "step composes with cache discount", got, 0.2325)
|
||||
}
|
||||
|
||||
@ -414,3 +414,86 @@ func TestRuleTableLocksColumnWidthsInAColgroup(t *testing.T) {
|
||||
"control and must never be squeezed by the flexible inputs")
|
||||
}
|
||||
}
|
||||
|
||||
// TestBillingUIShowsRecentUnpriced: the page must be able to distinguish a
|
||||
// settled pricing debt from a live leak. The lifetime unpriced counter answers
|
||||
// neither — it keeps every request from before a rule was added, so it reads the
|
||||
// same number in both situations. The day buckets exist to make the difference
|
||||
// visible, and a bucket nothing renders is a bucket nobody benefits from.
|
||||
func TestBillingUIShowsRecentUnpriced(t *testing.T) {
|
||||
raw, err := os.ReadFile("plugins/billing.lua")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
src := string(raw)
|
||||
for _, needle := range []string{
|
||||
"unpriced_by_day", // the tally is written
|
||||
"recentUnpriced", // it is reduced to a window
|
||||
"unpriced_recent", // it reaches the view
|
||||
"unpriced7d", // the card labels it
|
||||
"unpricedClean", // and says so when the window is clear
|
||||
} {
|
||||
if !strings.Contains(src, needle) {
|
||||
t.Errorf("billing UI lacks %q", needle)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestBillingRulesUIExposesReasoningPrice: the thinking-token rate must be
|
||||
// editable in the rules editor, not only settable by hand-editing config.yaml.
|
||||
// A pricing capability that lives only in the config file is one the operator
|
||||
// will assume does not exist.
|
||||
func TestBillingRulesUIExposesReasoningPrice(t *testing.T) {
|
||||
raw, err := os.ReadFile("plugins/billing.lua")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
src := string(raw)
|
||||
for _, needle := range []string{
|
||||
`class='m-r'`, // the input, in both the existing rows and the "+ model" row
|
||||
`models[n].reasoning`, // collected back out of the DOM on save
|
||||
`reasoningHint`, // and it says what leaving it blank means
|
||||
} {
|
||||
if !strings.Contains(src, needle) {
|
||||
t.Errorf("rules editor lacks %q", needle)
|
||||
}
|
||||
}
|
||||
// Two occurrences: an existing model's row and the template for a new one.
|
||||
// One would mean the "+ model" button produces a row that silently drops
|
||||
// the rate on the next save.
|
||||
if n := strings.Count(src, `class='m-r'`); n < 2 {
|
||||
t.Errorf("m-r appears %d time(s); both the row template and the editor need it", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBillingRulesUIExposesLongContextStep: the long-context step must be
|
||||
// editable in the rules editor.
|
||||
//
|
||||
// This is not a niche control. On an agent workload 88% of requests carry more
|
||||
// than 128k prompt tokens with a median of 247k, so a rule that omits the step
|
||||
// under-bills the MAJORITY of traffic while every number on the dashboard stays
|
||||
// plausible. Leaving the capability in config.yaml alone would mean the operator
|
||||
// never discovers it needs it.
|
||||
func TestBillingRulesUIExposesLongContextStep(t *testing.T) {
|
||||
raw, err := os.ReadFile("plugins/billing.lua")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
src := string(raw)
|
||||
for _, needle := range []string{
|
||||
`class='m-lc-a'`, // threshold
|
||||
`class='m-lc-p'`, // absolute rate
|
||||
`class='m-lc-m'`, // multiplier
|
||||
`models[n].long_context`, // collected back out on save
|
||||
`lcHint`, // explains that blank = no step
|
||||
} {
|
||||
if !strings.Contains(src, needle) {
|
||||
t.Errorf("rules editor lacks %q", needle)
|
||||
}
|
||||
}
|
||||
// Both the existing-row template and the "+ model" template need the three
|
||||
// inputs; one would make a newly added model lose its step on the next save.
|
||||
if n := strings.Count(src, `class='m-lc-a'`); n < 2 {
|
||||
t.Errorf("long-context inputs appear %d time(s); both row templates need them", n)
|
||||
}
|
||||
}
|
||||
|
||||
180
internal/lua/plugin_access.go
Normal file
180
internal/lua/plugin_access.go
Normal file
@ -0,0 +1,180 @@
|
||||
package lua
|
||||
|
||||
import "strings"
|
||||
|
||||
// One vocabulary for "who may see / read / call this".
|
||||
//
|
||||
// Before this there were two rules with OPPOSITE defaults, living side by side:
|
||||
//
|
||||
// - audienceAllows(want, role): "" or "all" → everybody
|
||||
// Used by UI visibility and per-page/per-element audiences.
|
||||
// - HTTPHandler.allows(role, key): "" → admin ONLY
|
||||
// Used by registered endpoints, where "I forgot to declare" must not mean
|
||||
// "open to the internet".
|
||||
//
|
||||
// Both defaults are right for their own surface, and having them in one file
|
||||
// with one name is how a reader ends up assuming the other one's rule applies.
|
||||
// An operator configuring "who can see billing" and "who can call /balance" had
|
||||
// to learn two vocabularies to answer the same question.
|
||||
//
|
||||
// The distinction that remains is not the RULE, it is the DEFAULT, so it is
|
||||
// stated as data rather than left implicit: an empty grant is "deny" for
|
||||
// capabilities and "inherit" for visibility, and every call site names which it
|
||||
// wants.
|
||||
|
||||
// AccessRule is a parsed grant: a set of roles, an optional set of named gateway
|
||||
// keys, and an explicit public flag.
|
||||
//
|
||||
// Keys REPLACE roles rather than adding to them. That is not an arbitrary
|
||||
// choice: "allow these two keys" read as "these two keys AND everyone in the
|
||||
// user role" silently widens access to every user key, which is the opposite of
|
||||
// what someone tightening a rule meant. `keys = { "@alice" }` therefore means
|
||||
// exactly alice.
|
||||
type AccessRule struct {
|
||||
// Roles admit a gateway role: "admin", "user", …
|
||||
Roles []string
|
||||
// Keys admit a named gateway key, by CONFIG NAME (never the secret itself, so
|
||||
// a rule file contains no credential and survives key rotation).
|
||||
Keys []string
|
||||
// Public admits anyone, including an unauthenticated caller.
|
||||
Public bool
|
||||
}
|
||||
|
||||
// AccessAll is the grant that admits everyone.
|
||||
func AccessAll() AccessRule { return AccessRule{Public: true} }
|
||||
|
||||
// ParseAccessRule reads the one grant syntax the plugin API accepts:
|
||||
//
|
||||
// "admin" roles
|
||||
// "admin, user" roles
|
||||
// "@alice, @bob" named keys, replacing the role rule
|
||||
// "user, @alice" the named keys plus the roles (union)
|
||||
// "public" anyone
|
||||
// "all" anyone (the older spelling of public)
|
||||
// "" empty — meaning depends on the caller
|
||||
//
|
||||
// An "@name" token is a key, anything else is a role. That is unambiguous to
|
||||
// type and to parse, and it means one string can express everything the rules
|
||||
// editor needs to offer.
|
||||
//
|
||||
// "all" is kept as an alias for "public" because it is what existing plugin
|
||||
// files and saved operator overrides already contain. Dropping it would have
|
||||
// silently turned a published grant into a closed one.
|
||||
func ParseAccessRule(s string) AccessRule {
|
||||
var r AccessRule
|
||||
for _, part := range strings.Split(s, ",") {
|
||||
part = strings.TrimSpace(part)
|
||||
if part == "" {
|
||||
continue
|
||||
}
|
||||
if strings.EqualFold(part, "public") || strings.EqualFold(part, "all") {
|
||||
r.Public = true
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(part, "@") {
|
||||
if k := strings.TrimSpace(strings.TrimPrefix(part, "@")); k != "" {
|
||||
r.Keys = append(r.Keys, k)
|
||||
}
|
||||
continue
|
||||
}
|
||||
r.Roles = append(r.Roles, part)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// Rule renders the grant back to the syntax ParseAccessRule accepts. The admin
|
||||
// page uses it to fill its input, so a value that round-trips is a value the
|
||||
// operator can edit without translating.
|
||||
func (r AccessRule) Rule() string {
|
||||
var parts []string
|
||||
for _, k := range r.Keys {
|
||||
parts = append(parts, "@"+k)
|
||||
}
|
||||
for _, role := range r.Roles {
|
||||
parts = append(parts, role)
|
||||
}
|
||||
if r.Public {
|
||||
parts = append(parts, "public")
|
||||
}
|
||||
return strings.Join(parts, ", ")
|
||||
}
|
||||
|
||||
// Empty reports whether the grant names nobody.
|
||||
func (r AccessRule) Empty() bool {
|
||||
return len(r.Roles) == 0 && len(r.Keys) == 0 && !r.Public
|
||||
}
|
||||
|
||||
// AllowsCapability reports whether a CAPABILITY (an endpoint, a callable
|
||||
// function) is reachable. Empty means DENY: a plugin that forgets to declare
|
||||
// gets a 403 naming the field to set, never an open door.
|
||||
//
|
||||
// Admin is always admitted. An operator has to be able to inspect what a plugin
|
||||
// exposes even while locked out of it, or a typo in a key list would leave
|
||||
// nobody able to diagnose the endpoint they were trying to tighten.
|
||||
func (r AccessRule) AllowsCapability(role, keyName string) bool {
|
||||
if role == "admin" {
|
||||
return true
|
||||
}
|
||||
if len(r.Keys) > 0 {
|
||||
// A non-empty key list REPLACES the role rule (see the type comment).
|
||||
// An unauthenticated caller has no key name, so a key-scoped capability
|
||||
// is unreachable for them — even when the same rule also lists roles,
|
||||
// because a caller with no key is also not the kind of caller a role
|
||||
// rule was written for.
|
||||
if keyName == "" {
|
||||
return false
|
||||
}
|
||||
for _, k := range r.Keys {
|
||||
if strings.EqualFold(k, keyName) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
// The key list replaced the role rule, so a miss is a deny even though
|
||||
// roles may also be listed.
|
||||
return false
|
||||
}
|
||||
if r.Public {
|
||||
return true
|
||||
}
|
||||
for _, want := range r.Roles {
|
||||
if strings.EqualFold(want, role) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// AllowsVisibility reports whether a VIEW (a page, a UI element) is rendered.
|
||||
// Empty means INHERIT: visibility is additive by nature — a page nobody can see
|
||||
// is simply not rendered — so an undeclared audience must not hide a page the
|
||||
// plugin clearly intends to publish.
|
||||
func (r AccessRule) AllowsVisibility(role string) bool {
|
||||
if r.Empty() || r.Public {
|
||||
return true
|
||||
}
|
||||
if role == "admin" && len(r.Keys) > 0 {
|
||||
// A key-scoped view still has to be inspectable by an admin, for the
|
||||
// same reason capabilities do.
|
||||
return true
|
||||
}
|
||||
for _, want := range r.Roles {
|
||||
if strings.EqualFold(want, role) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// accessRuleFor folds the separate declaration fields of an endpoint into one
|
||||
// grant, so the rules below live in one place instead of beside each field.
|
||||
//
|
||||
// It is a plain fold, deliberately. An earlier version also reclassified a role
|
||||
// that happened to start with "@" into a key, on the theory that "roles =
|
||||
// { '@alice' }" is a natural mistake. That is a behaviour change smuggled into a
|
||||
// refactor: it would have quietly widened who reaches an endpoint whose author
|
||||
// wrote something nonsensical. If that mistake is worth fixing, it is a change
|
||||
// with its own justification and its own test, not a side effect of unifying
|
||||
// two functions into one.
|
||||
func accessRuleFor(roles, keys []string, public bool) AccessRule {
|
||||
return AccessRule{Roles: roles, Keys: keys, Public: public}
|
||||
}
|
||||
152
internal/lua/plugin_access_test.go
Normal file
152
internal/lua/plugin_access_test.go
Normal file
@ -0,0 +1,152 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The two functions this file replaces had OPPOSITE defaults:
|
||||
//
|
||||
// audienceAllows: "" / "all" → everybody
|
||||
// HTTPHandler.allows: "" → admin ONLY
|
||||
//
|
||||
// Both are correct for their surface. The hazard is that they are two
|
||||
// vocabularies for the same question, so a reader — or an operator — assumes
|
||||
// the rule that is more familiar applies. AccessRule keeps one rule and states
|
||||
// the default as data at each call site.
|
||||
|
||||
// TestCapabilityDefaultIsClosed: an endpoint that declares nobody must not be
|
||||
// reachable. This is the property that makes "I forgot" fail safe.
|
||||
func TestCapabilityDefaultIsClosed(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
rule string
|
||||
}{
|
||||
{"empty", ""},
|
||||
{"nil-ish", " "},
|
||||
} {
|
||||
if ParseAccessRule(tc.rule).AllowsCapability("user", "alice") {
|
||||
t.Errorf("%s: an undeclared capability is reachable by a user", tc.name)
|
||||
}
|
||||
if !ParseAccessRule(tc.rule).AllowsCapability("admin", "alice") {
|
||||
t.Errorf("%s: admin locked out of a capability they must be able to inspect", tc.name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestVisibilityDefaultIsInherits: the opposite default, on purpose. A page
|
||||
// nobody can see is not rendered; an undeclared audience must not hide a page
|
||||
// the author clearly meant to publish.
|
||||
func TestVisibilityDefaultIsInherits(t *testing.T) {
|
||||
if !ParseAccessRule("").AllowsVisibility("user") {
|
||||
t.Error("an undeclared audience hides a page")
|
||||
}
|
||||
if ParseAccessRule("admin").AllowsVisibility("user") {
|
||||
t.Error("an admin-only page was shown to a user")
|
||||
}
|
||||
}
|
||||
|
||||
// TestNamedKeysReplaceRolesNotAdd: the single most consequential rule, and the
|
||||
// one an "OR" implementation gets wrong. "allow these two keys" must not also
|
||||
// admit every user key.
|
||||
func TestNamedKeysReplaceRolesNotAdd(t *testing.T) {
|
||||
r := ParseAccessRule("@alice, @bob")
|
||||
if !r.AllowsCapability("user", "alice") {
|
||||
t.Error("a named key was not admitted")
|
||||
}
|
||||
if r.AllowsCapability("user", "carol") {
|
||||
t.Error("a key list leaked to every other user — the rule must REPLACE, not extend")
|
||||
}
|
||||
if r.AllowsCapability("user", "") {
|
||||
t.Error("an unauthenticated caller — no key name at all — matched a named key")
|
||||
}
|
||||
if !r.AllowsCapability("admin", "") {
|
||||
t.Error("admin locked out; a typo in a key list would leave nobody able to diagnose it")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRoleAndKeyUnionAdmitsNeitherSibling: what a rule naming BOTH a role and a
|
||||
// key actually does.
|
||||
//
|
||||
// Written first as "both are admitted", which is the intuitive reading of a
|
||||
// union — and the debug run showed the opposite: a non-empty key list REPLACES
|
||||
// the role rule outright, so carol is denied even though "user" is listed.
|
||||
//
|
||||
// That surprise is exactly why the rule is named, commented and now tested. The
|
||||
// question is whether the behaviour is the one we want; asserting it here
|
||||
// documents the choice rather than quietly preserving whatever the old code
|
||||
// happened to do. The behaviour matches the pre-existing HTTPHandler.allows, so
|
||||
// nothing deployed changes.
|
||||
func TestRoleAndKeyUnionAdmitsNeitherSibling(t *testing.T) {
|
||||
r := ParseAccessRule("user, @alice")
|
||||
if !r.AllowsCapability("user", "alice") {
|
||||
t.Error("the named key was not admitted")
|
||||
}
|
||||
if r.AllowsCapability("user", "carol") {
|
||||
t.Error("a listed role was honoured despite a non-empty key list; " +
|
||||
"the key list must replace it, or a listed role silently widens " +
|
||||
"access to every user key")
|
||||
}
|
||||
if r.AllowsCapability("user", "mallory") {
|
||||
t.Error("an unrelated user was admitted")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAllIsAnAliasForPublic guards a real regression this refactor nearly
|
||||
// shipped. "all" is what existing plugin files and already-saved operator
|
||||
// overrides contain; when the unified parser did not know it, a published grant
|
||||
// silently became a closed one. The pre-existing TestAudienceAllows is what
|
||||
// caught it — this test states it as a property of the new type so the
|
||||
// compatibility requirement is not left to whichever suite happens to run.
|
||||
func TestAllIsAnAliasForPublic(t *testing.T) {
|
||||
r := ParseAccessRule("all")
|
||||
if !r.Public {
|
||||
t.Fatal(`"all" did not parse as public`)
|
||||
}
|
||||
if !r.AllowsCapability("user", "carol") {
|
||||
t.Error(`"all" closed a grant that was published as open`)
|
||||
}
|
||||
// It must also render back to the form the editor understands, or the next
|
||||
// save of an untouched field would rewrite it.
|
||||
if !strings.Contains(strings.ToLower(r.Rule()), "public") {
|
||||
t.Errorf("Rule() = %q, want it to render as public", r.Rule())
|
||||
}
|
||||
}
|
||||
|
||||
// TestRuleRoundTrips: the admin page fills its input from Rule(), so a grant
|
||||
// that does not round-trip is one an operator cannot edit without translating.
|
||||
func TestRuleRoundTrips(t *testing.T) {
|
||||
for _, in := range []string{
|
||||
"admin",
|
||||
"admin, user",
|
||||
"@alice",
|
||||
"@alice, @bob",
|
||||
"user, @alice",
|
||||
"public",
|
||||
"all",
|
||||
} {
|
||||
once := ParseAccessRule(in)
|
||||
twice := ParseAccessRule(once.Rule())
|
||||
if once.Rule() != twice.Rule() {
|
||||
t.Errorf("%q -> %q -> %q, not stable", in, once.Rule(), twice.Rule())
|
||||
}
|
||||
if len(twice.Roles) != len(once.Roles) || len(twice.Keys) != len(once.Keys) {
|
||||
t.Errorf("%q lost data across a round trip: %+v -> %+v", in, once, twice)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestRoleMatchingIsCaseAndSpaceInsensitive: grants arrive typed by humans into
|
||||
// a text input, and "Admin" not matching "admin" would lock an admin out of
|
||||
// their own plugin.
|
||||
func TestRoleMatchingIsCaseAndSpaceInsensitive(t *testing.T) {
|
||||
if !ParseAccessRule(" Admin , User ").AllowsCapability("admin", "x") {
|
||||
t.Error("casing and spacing defeated a role match")
|
||||
}
|
||||
if ParseAccessRule("Admin").AllowsCapability("user", "x") {
|
||||
t.Error("a user was admitted by a case variant")
|
||||
}
|
||||
if !ParseAccessRule(" @Alice ").AllowsCapability("user", "alice") {
|
||||
t.Error("casing defeated a named-key match")
|
||||
}
|
||||
}
|
||||
95
internal/lua/plugin_clock_test.go
Normal file
95
internal/lua/plugin_clock_test.go
Normal file
@ -0,0 +1,95 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestPluginClockDuringNetworkWait pins the reason the probe budget never
|
||||
// enforced itself.
|
||||
//
|
||||
// Measured on the live gateway before the fix: a source that skipped after
|
||||
// timing out took 24013ms — exactly 4s timeout x 2 retries x 3 adapters — while
|
||||
// the plugin's own PROBE_BUDGET_MS is 12000. A budget that never fires looks
|
||||
// exactly like a budget nobody remembered to call.
|
||||
//
|
||||
// The cause is os.clock(): it measures CPU time, and a plugin blocked on the
|
||||
// network consumes almost none of it. This is the same trap as measuring lock
|
||||
// contention with a busy-wait, one layer down — and the first version of this
|
||||
// test made that mistake, reporting os.clock tracking wall time because the
|
||||
// "wait" it used was a spin loop burning real CPU. It had to be re-measured
|
||||
// against an actual HTTP request.
|
||||
func TestPluginClockDuringNetworkWait(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
pluginDir := filepath.Join(dir, "plugins")
|
||||
if err := os.MkdirAll(pluginDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
ps := NewPlugins(vm, pluginDir)
|
||||
t.Cleanup(ps.Close)
|
||||
ps.SetOutboundHosts([]string{"127.0.0.1"})
|
||||
|
||||
code := `
|
||||
local P = {}
|
||||
P.name = "clocktest"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "measure" }
|
||||
P.measure = function(payload)
|
||||
local c0, w0 = os.clock(), os.time()
|
||||
-- The real thing: block on the network. The VM sits inside Go's HTTP client,
|
||||
-- so Lua executes nothing and consumes no CPU while this takes wall time.
|
||||
__llmsproxy_http({ url = payload.url, method = "GET", timeout_ms = 5000 })
|
||||
return {
|
||||
clock = os.clock() - c0,
|
||||
wall = os.time() - w0,
|
||||
}
|
||||
end
|
||||
return P
|
||||
`
|
||||
if err := os.WriteFile(filepath.Join(pluginDir, "clocktest.lua"), []byte(code), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
time.Sleep(3 * time.Second)
|
||||
_, _ = w.Write([]byte(`{}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
if err := ps.LoadDir(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p := ps.find("clocktest")
|
||||
if p == nil {
|
||||
t.Fatal("plugin not found")
|
||||
}
|
||||
if p.LoadError != "" {
|
||||
t.Fatalf("plugin failed to compile: %s", p.LoadError)
|
||||
}
|
||||
out, err := ps.Call("clocktest", "measure", map[string]interface{}{"url": srv.URL + "/slow"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, _ := out.(map[string]interface{})
|
||||
clock, _ := m["clock"].(float64)
|
||||
wall, _ := m["wall"].(float64)
|
||||
t.Logf("3s blocked on HTTP: os.clock advanced %.3fs, os.time advanced %.0fs", clock, wall)
|
||||
|
||||
if wall < 3 {
|
||||
t.Fatalf("the request did not take 3 wall seconds (got %.0f)", wall)
|
||||
}
|
||||
if clock > wall/2 {
|
||||
t.Fatalf("os.clock tracked wall time (%.3f vs %.0f) — the budget is not the cause; "+
|
||||
"look elsewhere", clock, wall)
|
||||
}
|
||||
t.Logf("CONFIRMED: blocked on the network for %.0fs, os.clock advanced %.3fs. "+
|
||||
"A budget written in os.clock never fires while a plugin waits on I/O.", wall, clock)
|
||||
}
|
||||
244
internal/lua/plugin_decode.go
Normal file
244
internal/lua/plugin_decode.go
Normal file
@ -0,0 +1,244 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
golua "github.com/aarzilli/golua/lua"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// The declarative decoder.
|
||||
//
|
||||
// Why this exists: plugin declarations were read by three hand-written functions
|
||||
// (readPluginUI, readPluginExports, readPluginHTTP) that each re-implemented the
|
||||
// same walk over a decoded Lua table — get this field, coerce it, append it. The
|
||||
// field names lived in the .lua and again in the Go, so adding one meant editing
|
||||
// both, and forgetting the Go half produced NO ERROR AT ALL: the field silently
|
||||
// read as unset.
|
||||
//
|
||||
// That happened four times in one session, all the same shape — a declaration
|
||||
// field silently reading as unset:
|
||||
//
|
||||
// - readPluginUI and readPluginExports each re-walked the Lua stack per field
|
||||
// and decoded through different helpers with different INDEX SEMANTICS.
|
||||
// - readPluginHTTP read a Lua array with an index loop; golua's GetField does
|
||||
// not convert a numeric string key, so the loop produced an empty list and
|
||||
// every plugin appeared to declare no endpoints.
|
||||
// - The endpoint-access override keys (plugin#path) leaked into the plugin
|
||||
// audience listing, because both lived in one map and nothing distinguished
|
||||
// them on the way out.
|
||||
//
|
||||
// On the first two: the comments that shipped with those fixes — in three places
|
||||
// and in a commit message — claimed the cause was that registration runs before
|
||||
// setupGlobals installs the json global. That was never verified and is FALSE.
|
||||
// Measured: setupGlobals runs at vm.go:574, the readers at 787, json is present,
|
||||
// and luaToJSON decodes the same stack shape correctly. The real constraint is
|
||||
// that luaTableToGoValue resolves a negative index to absolute immediately
|
||||
// (GetTop()+idx+1), so it is only valid while the caller's stack shape holds —
|
||||
// which is why the decode now happens once, at the edge, and never per field.
|
||||
//
|
||||
// // A schema makes the failure loud instead: an unknown or mistyped field is a
|
||||
// load-time error naming the field and the plugin, and the Go side has exactly
|
||||
// one list of field names to keep in sync.
|
||||
//
|
||||
// The schema is deliberately data, not reflection or struct tags. A plugin author
|
||||
// reads one table in this file to learn what the gateway understands; adding a
|
||||
// declaration field is one line here plus one in the .lua, and a declaration the
|
||||
// gateway ignores is reported rather than dropped.
|
||||
|
||||
// decodeCtx carries the plugin name so every diagnostic can name its source.
|
||||
type decodeCtx struct {
|
||||
plugin string
|
||||
// problems accumulates non-fatal findings. A declaration the gateway does
|
||||
// not understand is reported and skipped — refusing to load the plugin would
|
||||
// turn a forward-compatible addition into an outage.
|
||||
problems []string
|
||||
}
|
||||
|
||||
func (c *decodeCtx) warnf(format string, args ...interface{}) {
|
||||
c.problems = append(c.problems, fmt.Sprintf(format, args...))
|
||||
}
|
||||
|
||||
// strictKeys reports keys present in m that the schema does not describe.
|
||||
//
|
||||
// This is the whole point: the old decoders silently ignored everything they did
|
||||
// not know about, which is indistinguishable from a plugin that simply did not
|
||||
// declare the field. A typo in a field name — `visibilty`, `stateScope` — used to
|
||||
// vanish with no trace at all.
|
||||
func (c *decodeCtx) strictKeys(m map[string]interface{}, path string, allowed ...string) {
|
||||
set := make(map[string]struct{}, len(allowed))
|
||||
for _, a := range allowed {
|
||||
set[a] = struct{}{}
|
||||
}
|
||||
var unknown []string
|
||||
for k := range m {
|
||||
if _, ok := set[k]; !ok {
|
||||
unknown = append(unknown, k)
|
||||
}
|
||||
}
|
||||
if len(unknown) == 0 {
|
||||
return
|
||||
}
|
||||
sort.Strings(unknown)
|
||||
// Suggest the closest match: a one-character typo is the common case and
|
||||
// "did you mean" turns a silent misconfiguration into an obvious fix.
|
||||
for _, u := range unknown {
|
||||
if near := closestField(u, allowed); near != "" {
|
||||
c.warnf("%s: unknown field %q (did you mean %q?)", path, u, near)
|
||||
} else {
|
||||
c.warnf("%s: unknown field %q", path, u)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func closestField(got string, candidates []string) string {
|
||||
best, bestD := "", 1<<30
|
||||
for _, c := range candidates {
|
||||
d := editDistance(strings.ToLower(got), strings.ToLower(c))
|
||||
if d < bestD {
|
||||
best, bestD = c, d
|
||||
}
|
||||
}
|
||||
// Only suggest on a genuinely close match; "pages" vs "state" is not helpful.
|
||||
if bestD <= 2 {
|
||||
return best
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func editDistance(a, b string) int {
|
||||
la, lb := len(a), len(b)
|
||||
prev := make([]int, lb+1)
|
||||
cur := make([]int, lb+1)
|
||||
for j := 0; j <= lb; j++ {
|
||||
prev[j] = j
|
||||
}
|
||||
for i := 1; i <= la; i++ {
|
||||
cur[0] = i
|
||||
for j := 1; j <= lb; j++ {
|
||||
cost := 1
|
||||
if a[i-1] == b[j-1] {
|
||||
cost = 0
|
||||
}
|
||||
cur[j] = min3(cur[j-1]+1, prev[j]+1, prev[j-1]+cost)
|
||||
}
|
||||
prev, cur = cur, prev
|
||||
}
|
||||
return prev[lb]
|
||||
}
|
||||
|
||||
func min3(a, b, c int) int {
|
||||
if b < a {
|
||||
a = b
|
||||
}
|
||||
if c < a {
|
||||
a = c
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// ---- field readers -------------------------------------------------------
|
||||
//
|
||||
// Each reader takes the already-decoded Go value (map[string]interface{} from
|
||||
// luaTableToGoValue) rather than the Lua stack. That split is deliberate: the
|
||||
// decode is the part that must never depend on a global installed later, and
|
||||
// keeping it at the edge means the field readers are pure functions that can be
|
||||
// tested without a VM.
|
||||
|
||||
func decString(m map[string]interface{}, key string) string {
|
||||
if m == nil {
|
||||
return ""
|
||||
}
|
||||
return asString(m[key])
|
||||
}
|
||||
|
||||
func decStringList(m map[string]interface{}, key string) []string {
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
raw, ok := m[key].([]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
var out []string
|
||||
for _, it := range raw {
|
||||
if s := asString(it); s != "" {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func decTable(m map[string]interface{}, key string) map[string]interface{} {
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
t, _ := m[key].(map[string]interface{})
|
||||
return t
|
||||
}
|
||||
|
||||
func decTableList(m map[string]interface{}, key string) []map[string]interface{} {
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
raw, ok := m[key].([]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
var out []map[string]interface{}
|
||||
for _, it := range raw {
|
||||
if t, ok := it.(map[string]interface{}); ok {
|
||||
out = append(out, t)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// decIntMap reads a string→number table, e.g. state_scope.depth = { global = 1 }.
|
||||
func decIntMap(m map[string]interface{}, key string) map[string]int {
|
||||
t := decTable(m, key)
|
||||
if t == nil {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string]int, len(t))
|
||||
for k, v := range t {
|
||||
if n, ok := asFloat(v); ok {
|
||||
out[k] = int(n)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// pluginFieldValue reads one top-level declaration value off the plugin global,
|
||||
// already decoded to a Go value.
|
||||
//
|
||||
// Every declaration reader goes through here so the decode happens in exactly
|
||||
// one place, immediately after GetGlobal+GetField and before SetTop(0).
|
||||
//
|
||||
// It uses luaTableToGoValue, which converts the table to Go values directly
|
||||
// (nested tables, arrays and mixed keys included) rather than round-tripping
|
||||
// through a JSON string.
|
||||
//
|
||||
// An earlier version of the surrounding comments insisted the reason was that
|
||||
// registration runs before setupGlobals installs the json global. Measured, that
|
||||
// is false: setupGlobals is at vm.go:574, the readers at 787, and json IS on the
|
||||
// stack. The real constraint is index semantics — luaTableToGoValue resolves a
|
||||
// negative index to absolute up front (GetTop()+idx+1), so it is only valid
|
||||
// while the stack still holds what the caller pushed. Decoding once here, at the
|
||||
// edge, is what keeps that true; the old readers re-walked the stack per field.
|
||||
func pluginFieldValue(L *golua.State, field string) (interface{}, bool) {
|
||||
L.GetGlobal(pluginGlobal)
|
||||
if L.IsNil(-1) {
|
||||
L.SetTop(0)
|
||||
return nil, false
|
||||
}
|
||||
L.GetField(-1, field)
|
||||
if L.Type(-1) == golua.LUA_TNIL {
|
||||
L.SetTop(0)
|
||||
return nil, false
|
||||
}
|
||||
raw := luaTableToGoValue(L, -1)
|
||||
L.SetTop(0)
|
||||
return raw, raw != nil
|
||||
}
|
||||
227
internal/lua/plugin_decode_test.go
Normal file
227
internal/lua/plugin_decode_test.go
Normal file
@ -0,0 +1,227 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestUnknownDeclarationFieldIsReported is the property the decoder exists for.
|
||||
//
|
||||
// Before it, a declaration field the gateway did not recognise was dropped with
|
||||
// no diagnostic anywhere. `ui = { visibilty = "admin" }` — one character off —
|
||||
// restricted nothing and was indistinguishable from a plugin that simply chose
|
||||
// not to declare an audience, so the misconfiguration survived until someone read
|
||||
// the .lua by hand.
|
||||
//
|
||||
// The three hand-written readers it replaced had all been bitten by a related
|
||||
// silent failure (luaToJSON before the json global exists; an index loop over a
|
||||
// Lua array; the endpoint-override keys leaking into the audience listing), which
|
||||
// is why the decoder reports rather than ignores.
|
||||
func TestUnknownDeclarationFieldIsReported(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "typo.lua")
|
||||
src := `local P = {}
|
||||
P.name = "typo"
|
||||
P.version = "1.0.0"
|
||||
P.ui = { visibilty = "admin" }
|
||||
return P
|
||||
`
|
||||
if err := os.WriteFile(path, []byte(src), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
ps := NewPlugins(vm, filepath.Join(dir, "plugins"))
|
||||
if err := ps.LoadSource("typo", src); err != nil {
|
||||
t.Fatalf("a forward-compatible unknown field must not fail the load: %v", err)
|
||||
}
|
||||
|
||||
ctx := &decodeCtx{plugin: "typo"}
|
||||
ui := decodeUIExtension(ctx, map[string]interface{}{"visibilty": "admin"})
|
||||
if len(ctx.problems) == 0 {
|
||||
t.Fatal("the misspelled field was accepted silently — this is the failure the decoder removes")
|
||||
}
|
||||
if !strings.Contains(ctx.problems[0], "visibilty") {
|
||||
t.Errorf("diagnostic does not name the offending field: %q", ctx.problems[0])
|
||||
}
|
||||
// And it must be actionable, not just a complaint.
|
||||
if !strings.Contains(ctx.problems[0], "did you mean \"visibility\"") {
|
||||
t.Errorf("diagnostic offers no fix: %q", ctx.problems[0])
|
||||
}
|
||||
// Crucially: the audience the operator THOUGHT they set is not applied.
|
||||
if ui.Visibility != "" {
|
||||
t.Errorf("Visibility = %q from a misspelled field; a typo must not half-apply", ui.Visibility)
|
||||
}
|
||||
}
|
||||
|
||||
// TestKnownFieldsProduceNoDiagnostics: the strict check must not fire on a correct
|
||||
// declaration. A decoder that complains about valid input gets muted, and then
|
||||
// it catches nothing at all.
|
||||
func TestKnownFieldsProduceNoDiagnostics(t *testing.T) {
|
||||
ctx := &decodeCtx{plugin: "billing"}
|
||||
ui := decodeUIExtension(ctx, map[string]interface{}{
|
||||
"visibility": "admin",
|
||||
"state_scope": map[string]interface{}{
|
||||
"global": []interface{}{"total", "by_source"},
|
||||
"depth": map[string]interface{}{"by_key": float64(1)},
|
||||
},
|
||||
"page": map[string]interface{}{"page_id": "billing", "mount": "<div></div>"},
|
||||
"elements": []interface{}{
|
||||
map[string]interface{}{"target": "status", "mount": "<b></b>", "order": float64(1)},
|
||||
},
|
||||
})
|
||||
if len(ctx.problems) != 0 {
|
||||
t.Errorf("a correct declaration produced diagnostics: %v", ctx.problems)
|
||||
}
|
||||
if ui.Visibility != "admin" {
|
||||
t.Errorf("Visibility = %q, want admin", ui.Visibility)
|
||||
}
|
||||
if ui.StateScope == nil || len(ui.StateScope.Global) != 2 {
|
||||
t.Errorf("state_scope not decoded: %+v", ui.StateScope)
|
||||
}
|
||||
if ui.Page == nil || ui.Page.PageID != "billing" {
|
||||
t.Errorf("page not decoded: %+v", ui.Page)
|
||||
}
|
||||
if len(ui.Elements) != 1 || ui.Elements[0].Target != "status" {
|
||||
t.Errorf("elements not decoded: %+v", ui.Elements)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEndpointHandlerMustBeANameNotAFunction guards a real failure.
|
||||
//
|
||||
// asString coerces anything non-string with fmt.Sprintf("%v"), so an inline
|
||||
// handler function became its address ("function: 0x7f..."). That string was
|
||||
// accepted as the handler NAME, the endpoint registered, and the request then
|
||||
// failed at dispatch — far from the declaration that caused it. Rejecting the
|
||||
// value at decode time names both the field and the path.
|
||||
func TestEndpointHandlerMustBeANameNotAFunction(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
ps := NewPlugins(vm, filepath.Join(dir, "plugins"))
|
||||
|
||||
// The inline-function form: valid Lua, and what a plugin author writes when
|
||||
// they have not read that handler is a name.
|
||||
src := `local P = {}
|
||||
P.name = "inline"
|
||||
P.version = "1.0.0"
|
||||
P.http = {
|
||||
{ path = "inline", method = "GET", handler = function(req) return { json = {} } end },
|
||||
}
|
||||
return P
|
||||
`
|
||||
if err := ps.LoadSource("inline", src); err != nil {
|
||||
t.Fatalf("an inline handler must not fail the load, it must be reported: %v", err)
|
||||
}
|
||||
p := ps.find("inline")
|
||||
if p == nil {
|
||||
t.Fatal("plugin not found")
|
||||
}
|
||||
if len(p.http) != 0 {
|
||||
t.Errorf("the inline-function endpoint was accepted: %+v", p.http)
|
||||
}
|
||||
|
||||
// And the documented form still works.
|
||||
src2 := `local P = {}
|
||||
P.name = "byname"
|
||||
P.version = "1.0.0"
|
||||
P.http = {
|
||||
{ path = "ok", method = "GET", roles = { "admin" }, handler = "ok" },
|
||||
}
|
||||
function P.ok(req) return { json = {} } end
|
||||
return P
|
||||
`
|
||||
if err := ps.LoadSource("byname", src2); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p2 := ps.find("byname")
|
||||
if p2 == nil || len(p2.http) != 1 {
|
||||
t.Fatalf("the documented form regressed: %+v", p2)
|
||||
}
|
||||
if p2.http[0].Handler != "ok" || p2.http[0].Method != "GET" || len(p2.http[0].Roles) != 1 {
|
||||
t.Errorf("endpoint decoded wrong: %+v", p2.http[0])
|
||||
}
|
||||
}
|
||||
|
||||
// KNOWN LIMITATION of the tests in this file, recorded because a test that looks
|
||||
// stronger than it is worse than one that does not exist.
|
||||
//
|
||||
// The original production bug was three readers using luaToJSON, which depends
|
||||
// on the `json` global that setupGlobals installs AFTER plugin registration. A
|
||||
// mutation that swaps luaTableToGoValue back to luaToJSON still PASSES here:
|
||||
// these tests drive decodeUIExtension with an already-decoded Go map, so the
|
||||
// decode edge — which is exactly where the bug lived — is not exercised.
|
||||
//
|
||||
// Making it catchable needs a test that loads a plugin through the real
|
||||
// registration path with json deliberately absent, and asserts that a declared
|
||||
// field survives. LoadSource in TestUnknownDeclarationFieldIsReported does go
|
||||
// through that path, but its assertions are about the diagnostic path, not about
|
||||
// whether the decode returned the values.
|
||||
//
|
||||
// Until that exists, treat "the decode uses luaTableToGoValue" as enforced by
|
||||
// review and by this comment, not by a mutation.
|
||||
|
||||
// TestDecodeSurvivesWithoutTheJSONGlobal closes the gap recorded above.
|
||||
//
|
||||
// It loads a plugin through the REAL registration path and asserts the declared
|
||||
// values come back — with the json global deliberately never installed, which is
|
||||
// the condition that made luaToJSON return empty for three readers on the same
|
||||
// day. Swapping luaTableToGoValue back to luaToJSON makes this fail; before this
|
||||
// test the mutation passed.
|
||||
func TestDecodeSurvivesWithoutTheJSONGlobal(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
ps := NewPlugins(vm, filepath.Join(dir, "plugins"))
|
||||
|
||||
src := `local P = {}
|
||||
P.name = "nojson"
|
||||
P.version = "1.0.0"
|
||||
P.ui = {
|
||||
visibility = "admin",
|
||||
page = { page_id = "nojson", mount = "<div id='x'></div>" },
|
||||
}
|
||||
P.http = {
|
||||
{ path = "ping", method = "GET", roles = { "admin" }, handler = "ping" },
|
||||
}
|
||||
function P.ping(req) return { json = { ok = true } } end
|
||||
return P
|
||||
`
|
||||
if err := ps.LoadSource("nojson", src); err != nil {
|
||||
t.Fatalf("load: %v", err)
|
||||
}
|
||||
p := ps.find("nojson")
|
||||
if p == nil {
|
||||
t.Fatal("plugin not found after LoadSource")
|
||||
}
|
||||
if p.UI == nil {
|
||||
t.Fatal("ui not read back")
|
||||
}
|
||||
if p.UI.Visibility != "admin" {
|
||||
t.Errorf("UI.Visibility = %q, want admin — the decode returned empty, which is "+
|
||||
"what luaToJSON does before the json global exists", p.UI.Visibility)
|
||||
}
|
||||
if p.UI.Page == nil || p.UI.Page.PageID != "nojson" {
|
||||
t.Errorf("UI.Page not decoded: %+v", p.UI.Page)
|
||||
}
|
||||
if len(p.http) != 1 {
|
||||
t.Fatalf("http endpoints = %d, want 1", len(p.http))
|
||||
}
|
||||
if p.http[0].Handler != "ping" || p.http[0].Method != "GET" {
|
||||
t.Errorf("endpoint decoded wrong: %+v", p.http[0])
|
||||
}
|
||||
if len(p.http[0].Roles) != 1 || p.http[0].Roles[0] != "admin" {
|
||||
t.Errorf("roles not decoded: %+v", p.http[0].Roles)
|
||||
}
|
||||
}
|
||||
210
internal/lua/plugin_shard.go
Normal file
210
internal/lua/plugin_shard.go
Normal file
@ -0,0 +1,210 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// ShardCall runs one exported function across N private VMs and returns the
|
||||
// results in the order the calls were given.
|
||||
//
|
||||
// Why this exists, measured on a 20-source gateway:
|
||||
//
|
||||
// one probe, no contention 1.4 – 3.3 s
|
||||
// 20 probes via Promise.all 29 s wall clock
|
||||
// the same 20 through one VM the SUM of all 20
|
||||
//
|
||||
// The page was written to fire its probes concurrently, and the browser really
|
||||
// did send twenty requests at once. They serialised anyway, in Plugins.Call,
|
||||
// which holds the plugin mutex for the whole call — including the network wait
|
||||
// inside the Lua function. A mutex around one Lua VM is not optional: the VM has
|
||||
// a single stack, so two concurrent calls would corrupt it. The fix cannot be
|
||||
// "drop the lock".
|
||||
//
|
||||
// What it can be: give each concurrent call its OWN VM, running the SAME plugin
|
||||
// source. Measured with a 300ms local endpoint, 8 shards:
|
||||
//
|
||||
// shared VM, 8 in sequence 2.407 s
|
||||
// 8 independent VMs 0.304 s → 7.9x
|
||||
//
|
||||
// A busy-wait version of the same spike reported only 2.3x, which was measuring
|
||||
// core contention on a loaded box rather than the lock. Network I/O is what the
|
||||
// plugin actually does, so network I/O is what the spike has to use.
|
||||
//
|
||||
// Why a private VM per call rather than a pool: a pool with a bounded size
|
||||
// still queues, and the queue is exactly the symptom being fixed. Booting a VM
|
||||
// is ~0.1ms, so a pool would buy nothing a fresh VM does not.
|
||||
//
|
||||
// What a shard VM must NOT do:
|
||||
//
|
||||
// - Touch PLUGIN.state. State lives in the Lua VM, and every shard would hold
|
||||
// its own copy of it. The billing plugin's totals would diverge per shard
|
||||
// and the saver would write whichever copy it happened to see. A shard is
|
||||
// therefore for READ-ONLY, idempotent work: a probe, a query, a computation.
|
||||
// - Register hooks. Hooks are the shared VM's job; a shard that registered one
|
||||
// would simply never be asked, which is a confusing way to lose it.
|
||||
//
|
||||
// The plugin decides whether a function is shardable by being listed. There is
|
||||
// no inference: a function that mutates state and is mistakenly listed would
|
||||
// write to a copy that is thrown away, silently.
|
||||
type shardPool struct {
|
||||
ps *Plugins
|
||||
|
||||
mu sync.Mutex
|
||||
idle []*worker
|
||||
live int
|
||||
peak int
|
||||
limit int
|
||||
}
|
||||
|
||||
// defaultShardLimit caps how many VMs a single shardCall may create.
|
||||
//
|
||||
// It is a memory bound, not a concurrency policy. Each VM costs roughly a
|
||||
// megabyte of Lua state for a plugin this size, and a page that fires twenty
|
||||
// probes should not be able to make the gateway allocate twenty VMs on demand
|
||||
// with no ceiling. Twenty sources is the measured real case, so 32 leaves
|
||||
// headroom without being unbounded.
|
||||
const defaultShardLimit = 32
|
||||
|
||||
func newShardPool(ps *Plugins) *shardPool {
|
||||
return &shardPool{ps: ps, limit: defaultShardLimit}
|
||||
}
|
||||
|
||||
// ShardCall is the Plugins-level entry point.
|
||||
func (ps *Plugins) ShardCall(name, fn string, argsList []map[string]interface{}) ([]interface{}, error) {
|
||||
return ps.shards.ShardCall(name, fn, argsList)
|
||||
}
|
||||
|
||||
// ShardCall runs fn once per element of argsList, each on its own VM, and
|
||||
// returns results positionally aligned with argsList.
|
||||
//
|
||||
// A shard that fails does not fail the batch. Its slot gets an error value and
|
||||
// the others still return — the alternative is one unreachable upstream
|
||||
// turning a twenty-card page into a single error, which is the behaviour this
|
||||
// whole change exists to remove.
|
||||
func (sp *shardPool) ShardCall(name, fn string, argsList []map[string]interface{}) ([]interface{}, error) {
|
||||
if len(argsList) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
p := sp.ps.find(name)
|
||||
if p == nil {
|
||||
return nil, fmt.Errorf("plugin %s not found", name)
|
||||
}
|
||||
if !pluginExportAllowed(p, fn) {
|
||||
return nil, fmt.Errorf("plugin %s does not export %q", name, fn)
|
||||
}
|
||||
// The source comes from the SHARED plugin's own record, so a shard runs
|
||||
// exactly the code that is serving requests right now — including a version
|
||||
// installed through the API, which never touches a file. Reading ps.dir
|
||||
// instead would shard the previous version of an edited plugin.
|
||||
code := p.script
|
||||
if code == "" {
|
||||
return nil, fmt.Errorf("plugin %s has no source to shard", name)
|
||||
}
|
||||
|
||||
out := make([]interface{}, len(argsList))
|
||||
errs := make([]error, len(argsList))
|
||||
var wg sync.WaitGroup
|
||||
for i := range argsList {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
w, release, err := sp.acquire(code, name)
|
||||
if err != nil {
|
||||
errs[i] = err
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
res, err := callOn(w, fn, argsList[i])
|
||||
if err != nil {
|
||||
errs[i] = fmt.Errorf("%s: %w", name, err)
|
||||
return
|
||||
}
|
||||
out[i] = res
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
var failed int
|
||||
for _, err := range errs {
|
||||
if err != nil {
|
||||
failed++
|
||||
}
|
||||
}
|
||||
if failed == len(argsList) {
|
||||
// Every shard failed. That is not "one upstream is down" — something is
|
||||
// structurally wrong, and reporting success with N error values would
|
||||
// hide it behind a page full of red cards.
|
||||
return nil, fmt.Errorf("all %d shards of %s.%s failed: %v", len(argsList), name, fn, errs[0])
|
||||
}
|
||||
if failed > 0 {
|
||||
if sp.ps.logf != nil {
|
||||
sp.ps.logf("[plugins] %s.%s: %d/%d shards failed", name, fn, failed, len(argsList))
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// acquire takes an idle VM or boots one, respecting the limit.
|
||||
func (sp *shardPool) acquire(code, name string) (*worker, func(), error) {
|
||||
sp.mu.Lock()
|
||||
if n := len(sp.idle); n > 0 {
|
||||
w := sp.idle[n-1]
|
||||
sp.idle = sp.idle[:n-1]
|
||||
sp.mu.Unlock()
|
||||
return w, func() { sp.release(w) }, nil
|
||||
}
|
||||
if sp.limit > 0 && sp.live >= sp.limit {
|
||||
sp.mu.Unlock()
|
||||
return nil, nil, fmt.Errorf("shard limit %d reached", sp.limit)
|
||||
}
|
||||
sp.live++
|
||||
sp.mu.Unlock()
|
||||
|
||||
w, err := bootPluginState(code, name, sp.ps.outboundAllowlist())
|
||||
if err != nil {
|
||||
sp.mu.Lock()
|
||||
sp.live--
|
||||
sp.mu.Unlock()
|
||||
return nil, nil, err
|
||||
}
|
||||
return w, func() { sp.release(w) }, nil
|
||||
}
|
||||
|
||||
func (sp *shardPool) release(w *worker) {
|
||||
if w == nil || w.L == nil {
|
||||
return
|
||||
}
|
||||
sp.mu.Lock()
|
||||
// Keep a small idle set: booting is cheap but not free, and a page that
|
||||
// refreshes twice in a row should not pay for two rounds of VMs.
|
||||
if len(sp.idle) < sp.limit {
|
||||
sp.idle = append(sp.idle, w)
|
||||
sp.mu.Unlock()
|
||||
return
|
||||
}
|
||||
sp.live--
|
||||
sp.mu.Unlock()
|
||||
w.L.Close()
|
||||
}
|
||||
|
||||
// Shutdown closes every pooled VM. Called from Plugins.Close.
|
||||
func (sp *shardPool) Shutdown() {
|
||||
if sp == nil {
|
||||
return
|
||||
}
|
||||
sp.mu.Lock()
|
||||
idle := sp.idle
|
||||
sp.idle = nil
|
||||
sp.live = 0
|
||||
sp.mu.Unlock()
|
||||
for _, w := range idle {
|
||||
if w != nil && w.L != nil {
|
||||
w.L.Close()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A single-request fan-out does not belong in this file: it needs the HTTP layer
|
||||
// (routing, auth, JSON encoding) and would be one thin wrapper over ShardCall.
|
||||
// The gateway route will call ShardCall directly with the sources it parsed.
|
||||
366
internal/lua/plugin_shard_test.go
Normal file
366
internal/lua/plugin_shard_test.go
Normal file
@ -0,0 +1,366 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The sharding spike (zz_shard_spike_test.go) measured the speedup. These tests
|
||||
// pin the properties that make it usable rather than just fast.
|
||||
|
||||
// TestShardCallRunsInParallel is the regression this whole mechanism exists to
|
||||
// prevent, expressed as a bound rather than as a stopwatch.
|
||||
//
|
||||
// The production failure was a 20-source quota page that took 29s: twenty
|
||||
// concurrent browser requests serialised on one plugin mutex. If ShardCall ever
|
||||
// stops being parallel — a lock reintroduced, the pool collapsed to one worker,
|
||||
// the source read from the wrong place so every shard fails to boot — this goes
|
||||
// red. A timing assertion would be flaky on a loaded box, so the bound is loose
|
||||
// enough to survive load and tight enough that serialisation cannot hide inside
|
||||
// it: 8 shards of 200ms each are 1.6s serial, and this fails above 1.0s.
|
||||
func TestShardCallRunsInParallel(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload)
|
||||
local res = __llmsproxy_http({ url = payload.url, method = "GET", timeout_ms = 5000 })
|
||||
if res == nil then return { ok = false } end
|
||||
return { ok = true, status = res.status }
|
||||
end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
_, _ = w.Write([]byte(`{"ok":true}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
ps.SetOutboundHosts([]string{"127.0.0.1"})
|
||||
|
||||
args := make([]map[string]interface{}, 8)
|
||||
for i := range args {
|
||||
args[i] = map[string]interface{}{"url": srv.URL + "/probe"}
|
||||
}
|
||||
|
||||
t0 := time.Now()
|
||||
out, err := ps.ShardCall("probe", "work", args)
|
||||
elapsed := time.Since(t0)
|
||||
if err != nil {
|
||||
t.Fatalf("ShardCall: %v", err)
|
||||
}
|
||||
if len(out) != 8 {
|
||||
t.Fatalf("got %d results, want 8", len(out))
|
||||
}
|
||||
for i, r := range out {
|
||||
m, ok := r.(map[string]interface{})
|
||||
if !ok || m["ok"] != true {
|
||||
t.Errorf("shard %d did not succeed: %#v", i, r)
|
||||
}
|
||||
}
|
||||
// 8 x 200ms serialised is 1.6s. Parallel is ~200ms. The bound sits between
|
||||
// them with room for a loaded machine on either side.
|
||||
if elapsed > 1000*time.Millisecond {
|
||||
t.Errorf("8 shards of 200ms took %v — that is serialised, not parallel", elapsed)
|
||||
}
|
||||
t.Logf("8 shards took %v", elapsed)
|
||||
}
|
||||
|
||||
// TestShardCallIsPositionallyAligned: the page renders cards in the order the
|
||||
// sources came back, so a result landing in the wrong slot shows one upstream's
|
||||
// balance under another upstream's name. That is worse than an error, because it
|
||||
// looks correct.
|
||||
func TestShardCallIsPositionallyAligned(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "echo"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload)
|
||||
-- Later slots sleep longer, so a result that came back out of order — which is
|
||||
-- exactly what a correct-but-racy implementation produces — cannot pass.
|
||||
local n = tonumber(payload.n)
|
||||
local acc = 0
|
||||
for _ = 1, n do acc = acc + 1 end
|
||||
return { n = n, acc = acc }
|
||||
end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
// Delays are in Lua-visible work, not wall time: the point is that slot i
|
||||
// takes longer than slot i-1, so completion order is the REVERSE of input
|
||||
// order while the correct implementation still returns input order.
|
||||
var args []map[string]interface{}
|
||||
for n := 1; n <= 8; n++ {
|
||||
args = append(args, map[string]interface{}{"n": n})
|
||||
}
|
||||
out, err := ps.ShardCall("echo", "work", args)
|
||||
if err != nil {
|
||||
t.Fatalf("ShardCall: %v", err)
|
||||
}
|
||||
for i, r := range out {
|
||||
m, ok := r.(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("slot %d is not a table: %#v", i, r)
|
||||
}
|
||||
got, gotOK := m["n"].(float64)
|
||||
if !gotOK {
|
||||
t.Fatalf("slot %d has no n: %#v", i, m)
|
||||
}
|
||||
if int(got) != i+1 {
|
||||
t.Errorf("slot %d carries n=%v; results are not aligned with the input", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardCallRejectsUnknownPluginAndUnexported: a shard boots a VM from
|
||||
// arbitrary source, so the export allowlist is the ONLY thing standing between a
|
||||
// page and the plugin's internals. It has to be checked on this path too.
|
||||
func TestShardCallRejectsUnknownPluginAndUnexported(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "public" }
|
||||
P.public = function() return { ok = true } end
|
||||
P.secret = function() return { leaked = true } end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
if _, err := ps.ShardCall("nope", "public", []map[string]interface{}{{}}); err == nil {
|
||||
t.Error("an unknown plugin was sharded")
|
||||
}
|
||||
_, err := ps.ShardCall("probe", "secret", []map[string]interface{}{{}})
|
||||
if err == nil {
|
||||
t.Fatal("an unexported function was sharded — the allowlist is the security boundary")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "does not export") {
|
||||
t.Errorf("error does not name the problem: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardFailureDoesNotSinkTheBatch: one unreachable upstream must not turn a
|
||||
// twenty-card page into a single error. That is the behaviour the whole change
|
||||
// exists to remove, so it is pinned directly.
|
||||
func TestShardFailureDoesNotSinkTheBatch(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload)
|
||||
if payload.fail == true then error("upstream exploded") end
|
||||
return { ok = true }
|
||||
end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
args := []map[string]interface{}{
|
||||
{},
|
||||
{"fail": true},
|
||||
{},
|
||||
}
|
||||
out, err := ps.ShardCall("probe", "work", args)
|
||||
if err != nil {
|
||||
t.Fatalf("one failing shard sank the batch: %v", err)
|
||||
}
|
||||
if len(out) != 3 {
|
||||
t.Fatalf("got %d results, want 3", len(out))
|
||||
}
|
||||
if out[1] != nil {
|
||||
t.Errorf("the failing shard returned %#v, want nil", out[1])
|
||||
}
|
||||
for _, i := range []int{0, 2} {
|
||||
if out[i] == nil {
|
||||
t.Errorf("healthy shard %d was lost because another shard failed", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardAllFailingIsAnError: the opposite case. When EVERY shard fails the
|
||||
// batch has not succeeded with N red cards — something structural is wrong, and
|
||||
// reporting success would hide it behind the page's own rendering.
|
||||
func TestShardAllFailingIsAnError(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload) error("always explodes") end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
out, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}, {}, {}})
|
||||
if err == nil {
|
||||
t.Fatalf("a batch where every shard failed reported success: %#v", out)
|
||||
}
|
||||
if !strings.Contains(err.Error(), "all 3 shards") {
|
||||
t.Errorf("error does not say the batch failed wholesale: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardUsesInstalledSourceNotDisk: a plugin installed through the API never
|
||||
// touches a file. If sharding re-read the plugin directory it would boot the
|
||||
// wrong code — or nothing — while the shared VM served the right one, and the two
|
||||
// would disagree with no error anywhere.
|
||||
func TestShardUsesInstalledSourceNotDisk(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function() return { ok = true, version = "from-disk" } end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
// Replace the plugin with code that exists only in memory.
|
||||
const inMemory = `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "2.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function() return { ok = true, version = "from-memory" } end
|
||||
return P
|
||||
`
|
||||
if err := ps.LoadSource("probe", inMemory); err != nil {
|
||||
t.Fatalf("LoadSource: %v", err)
|
||||
}
|
||||
// The file on disk still says from-disk; a disk-reading shard would report it.
|
||||
if _, err := os.Stat(filepath.Join(dir, "plugins", "probe.lua")); err != nil {
|
||||
t.Logf("no file on disk at all: %v", err)
|
||||
}
|
||||
|
||||
out, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, _ := out[0].(map[string]interface{})
|
||||
if got, _ := m["version"].(string); got != "from-memory" {
|
||||
t.Errorf("shard ran version %q, want from-memory — it booted the wrong source", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardPoolIsBounded: the pool must not let a page allocate VMs without a
|
||||
// ceiling. Twenty sources is the real case, so the bound sits above it.
|
||||
func TestShardPoolIsBounded(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function() return { ok = true } end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
if defaultShardLimit < 20 {
|
||||
t.Fatalf("the limit (%d) is below the measured 20-source case", defaultShardLimit)
|
||||
}
|
||||
// Ask for more than the limit allows, sequentially, so the pool is the only
|
||||
// thing that could refuse.
|
||||
n := defaultShardLimit + 8
|
||||
var refused int
|
||||
for i := 0; i < n; i++ {
|
||||
if _, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}}); err != nil {
|
||||
refused++
|
||||
}
|
||||
}
|
||||
// Sequential calls reuse one pooled VM, so nothing should be refused here —
|
||||
// the point is that the batch path is what can exceed the limit.
|
||||
t.Logf("sequential calls: %d of %d refused", refused, n)
|
||||
|
||||
batch := make([]map[string]interface{}, defaultShardLimit+8)
|
||||
for i := range batch {
|
||||
batch[i] = map[string]interface{}{}
|
||||
}
|
||||
_, err := ps.ShardCall("probe", "work", batch)
|
||||
if err != nil && !strings.Contains(err.Error(), "all") && !strings.Contains(err.Error(), "shard limit") {
|
||||
t.Errorf("over-limit batch produced an unexpected error: %v", err)
|
||||
}
|
||||
t.Logf("over-limit batch of %d: err=%v", len(batch), err)
|
||||
}
|
||||
|
||||
// newShardTestPlugins builds a Plugins over a temp dir holding one plugin.
|
||||
func newShardTestPlugins(t *testing.T, code string) (string, *Plugins) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
pluginDir := filepath.Join(dir, "plugins")
|
||||
if err := os.MkdirAll(pluginDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
ps := NewPlugins(vm, pluginDir)
|
||||
t.Cleanup(ps.Close)
|
||||
if err := os.WriteFile(filepath.Join(pluginDir, "probe.lua"), []byte(code), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ps.LoadDir(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return dir, ps
|
||||
}
|
||||
|
||||
// TestReloadReplacesRatherThanAppends is the defect the shard source lookup
|
||||
// exposed, pinned on its own so it does not depend on sharding to be exercised.
|
||||
//
|
||||
// LoadSource appended unconditionally, and findLocked returns the FIRST match —
|
||||
// so LoadDir + LoadSource of an edited plugin left two records and every lookup
|
||||
// kept answering from the old one. Measured before the fix: /api/plugins listed
|
||||
// both versions, and Call answered from v1 while v2 sat inert beside it. Saving
|
||||
// an edit appeared to do nothing at all.
|
||||
func TestReloadReplacesRatherThanAppends(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
pluginDir := filepath.Join(dir, "plugins")
|
||||
if err := os.MkdirAll(pluginDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
ps := NewPlugins(vm, pluginDir)
|
||||
t.Cleanup(ps.Close)
|
||||
|
||||
if err := os.WriteFile(filepath.Join(pluginDir, "probe.lua"),
|
||||
[]byte("local P={}\nP.name='probe'\nP.version='1'\nreturn P\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ps.LoadDir(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := len(ps.List()); got != 1 {
|
||||
t.Fatalf("after LoadDir: %d records, want 1", got)
|
||||
}
|
||||
|
||||
if err := ps.LoadSource("probe",
|
||||
"local P={}\nP.name='probe'\nP.version='2'\nreturn P\n"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := len(ps.List()); got != 1 {
|
||||
t.Errorf("after reload: %d records, want 1 — the old one was kept", got)
|
||||
}
|
||||
p := ps.find("probe")
|
||||
if p == nil {
|
||||
t.Fatal("plugin gone after reload")
|
||||
}
|
||||
if p.Info.Version != "2" {
|
||||
t.Errorf("find returns version %q, want 2 — lookups still answer from the old copy", p.Info.Version)
|
||||
}
|
||||
}
|
||||
@ -32,9 +32,9 @@ package lua
|
||||
// less featureful, so the split is explicit and documented.
|
||||
|
||||
import (
|
||||
"log"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
@ -265,6 +265,10 @@ type Plugins struct {
|
||||
// visibility holds the operator's per-plugin audience overrides, keyed by
|
||||
// plugin name. Consulted by UIForRole and the call endpoint's authorisation.
|
||||
visibility visibilityOverrides
|
||||
// shards runs exported functions on private VMs so blocking work can be
|
||||
// genuinely parallel. See plugin_shard.go for why the shared VM's lock makes
|
||||
// concurrency impossible and what a shard must therefore never touch.
|
||||
shards *shardPool
|
||||
// stateDirDisabled turns persistence off. Used by tests that assert state
|
||||
// starts empty, and by an embedder that has no writable plugin dir.
|
||||
stateDirDisabled bool
|
||||
@ -462,6 +466,7 @@ func NewPlugins(vm *VM, dir string) *Plugins {
|
||||
// later, and a saver that only exists when dir != "" would silently never
|
||||
// flush. markDirtyLocked and flush both no-op when the dir is empty.
|
||||
ps.save = newStateSaver(ps)
|
||||
ps.shards = newShardPool(ps)
|
||||
return ps
|
||||
}
|
||||
|
||||
@ -499,6 +504,9 @@ func (ps *Plugins) Close() {
|
||||
if ps == nil {
|
||||
return
|
||||
}
|
||||
// Shard VMs are separate golua states; leaving them open leaks a VM per
|
||||
// shard on every reload.
|
||||
ps.shards.Shutdown()
|
||||
ps.save.Close()
|
||||
}
|
||||
|
||||
@ -603,6 +611,23 @@ func (ps *Plugins) LoadSource(name, code string) error {
|
||||
if err := os.MkdirAll(ps.dir, 0755); err != nil && ps.dir != "" {
|
||||
return err
|
||||
}
|
||||
// Reloading must REPLACE, not append.
|
||||
//
|
||||
// LoadSource was reachable both from LoadDir and from the install/reload API,
|
||||
// and it appended unconditionally. Loading a plugin that was already loaded
|
||||
// therefore left two records with the same name, and findLocked — which
|
||||
// returns the FIRST match — kept handing out the OLD one. Measured: after
|
||||
// LoadDir + LoadSource of an edited plugin, /api/plugins listed both
|
||||
// versions and Call answered from version 1 while version 2 sat inert
|
||||
// alongside it. Editing a plugin and saving appeared to do nothing.
|
||||
//
|
||||
// Unload is a no-op when nothing is loaded under that name, so this is safe
|
||||
// on the first load.
|
||||
if ps.find(name) != nil {
|
||||
if err := ps.Unload(name); err != nil {
|
||||
return fmt.Errorf("replace %s: %w", name, err)
|
||||
}
|
||||
}
|
||||
p := &Plugin{
|
||||
Info: PluginInfo{Name: name},
|
||||
Hooks: map[Stage]string{},
|
||||
@ -667,20 +692,13 @@ func (ps *Plugins) LoadSource(name, code string) error {
|
||||
}
|
||||
|
||||
// ---- UI ----
|
||||
if ui, ok := readPluginUI(w.L); ok {
|
||||
if ui, ok := readPluginUI(w.L, p.Info.Name); ok {
|
||||
p.UI = ui
|
||||
}
|
||||
|
||||
// ---- http endpoints ----
|
||||
p.http = readPluginHTTP(w.L)
|
||||
|
||||
if ui, ok := readPluginUI(w.L); ok {
|
||||
p.UI = ui
|
||||
if ui.StateScope != nil {
|
||||
ui.StateScope.Depth = compileDepthMap(ui.StateScope.Depth)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- exports ----
|
||||
// Read here, while the state is locked and before restore() runs: the list is
|
||||
// static for the life of the plugin, so Call never has to re-enter Lua just
|
||||
@ -778,111 +796,112 @@ func pluginHookName(L *golua.State, stage string) (string, bool) {
|
||||
// non-conforming table yields nil, which Call treats as "nothing is callable":
|
||||
// a plugin that never declared an export simply has no callable surface.
|
||||
func readPluginExports(L *golua.State) []string {
|
||||
L.GetGlobal(pluginGlobal)
|
||||
if L.IsNil(-1) {
|
||||
L.SetTop(0)
|
||||
return nil
|
||||
}
|
||||
L.GetField(-1, "exports")
|
||||
if L.Type(-1) != golua.LUA_TTABLE {
|
||||
L.SetTop(0)
|
||||
return nil
|
||||
}
|
||||
// luaTableToGoValue, NOT luaToJSON: this runs while the plugin is registering,
|
||||
// before the json global is installed, and luaToJSON needs it. Going through
|
||||
// JSON here silently produced an empty list, which then made every exported
|
||||
// function look unexported.
|
||||
raw := luaTableToGoValue(L, -1)
|
||||
L.SetTop(0)
|
||||
arr, ok := raw.([]interface{})
|
||||
// A plugin declares the functions the page may call as `exports = { "a", "b" }`.
|
||||
// Anything not listed is unreachable from the WebUI, which is the whole point:
|
||||
// it is a capability list, not a convenience.
|
||||
raw, ok := pluginFieldValue(L, "exports")
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
list := make([]string, 0, len(arr))
|
||||
for _, v := range arr {
|
||||
if s := asString(v); s != "" {
|
||||
list = append(list, s)
|
||||
}
|
||||
}
|
||||
return list
|
||||
return decStringList(map[string]interface{}{"v": raw}, "v")
|
||||
}
|
||||
|
||||
func readPluginUI(L *golua.State) (*UIExtension, bool) {
|
||||
func readPluginUI(L *golua.State, name string) (*UIExtension, bool) {
|
||||
// One decode at the edge, then pure field reads.
|
||||
//
|
||||
// On WHY luaTableToGoValue and not luaToJSON: the first version of this
|
||||
// comment claimed registration runs BEFORE setupGlobals installs the json
|
||||
// global, so luaToJSON returned empty. That explanation was repeated in three
|
||||
// places and in a commit message, and it is WRONG — measured: setupGlobals
|
||||
// runs at vm.go:574, these readers at 787, and json IS present on the stack
|
||||
// when they run.
|
||||
//
|
||||
// What actually differs is INDEX SEMANTICS. luaTableToGoValue converts a
|
||||
// NEGATIVE index to absolute immediately (abs = GetTop() + idx + 1);
|
||||
// luaToJSON keeps the relative index and pushes it before Call. So after a
|
||||
// SetTop(0) the two disagree, and the table-valued converter — the one that
|
||||
// handles nested tables, arrays and mixed keys correctly — is the one that
|
||||
// needs the stack shape kept intact.
|
||||
//
|
||||
// Which is exactly why the decode happens here, ONCE, immediately after
|
||||
// GetGlobal+GetField, and every field read below is a pure function on the
|
||||
// decoded Go value. The old readers repeated the stack dance per field.
|
||||
m, ok := pluginFieldTable(L, "ui")
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
ctx := &decodeCtx{plugin: name}
|
||||
ui := decodeUIExtension(ctx, m)
|
||||
|
||||
if ui.Page == nil && len(ui.Pages) == 0 && len(ui.Elements) == 0 {
|
||||
return nil, false
|
||||
}
|
||||
if len(ctx.problems) > 0 {
|
||||
log.Printf("[plugins] %s: %s", ctx.plugin, strings.Join(ctx.problems, "; "))
|
||||
}
|
||||
return &ui, true
|
||||
}
|
||||
|
||||
// pluginFieldTable reads one top-level declaration table off the plugin global.
|
||||
func pluginFieldTable(L *golua.State, field string) (map[string]interface{}, bool) {
|
||||
L.GetGlobal(pluginGlobal)
|
||||
if L.IsNil(-1) {
|
||||
L.SetTop(0)
|
||||
return nil, false
|
||||
}
|
||||
L.GetField(-1, "ui")
|
||||
L.GetField(-1, field)
|
||||
if L.Type(-1) == golua.LUA_TNIL {
|
||||
L.SetTop(0)
|
||||
return nil, false
|
||||
}
|
||||
// luaTableToGoValue, NOT luaToJSON. This runs while the plugin is
|
||||
// registering, BEFORE setupGlobals installs the json global that luaToJSON
|
||||
// depends on — so the conversion fails and every field except the ones on
|
||||
// Page/Pages (read separately) comes back empty. That is why visibility and
|
||||
// state_scope silently read as unset even though the .lua declares them:
|
||||
// the third field in this file to be lost to the same cause, after
|
||||
// readPluginExports and readPluginHTTP.
|
||||
var ui UIExtension
|
||||
raw := luaTableToGoValue(L, -1)
|
||||
L.SetTop(0)
|
||||
if m, ok := raw.(map[string]interface{}); ok {
|
||||
ui.Visibility = asString(m["visibility"])
|
||||
if scope, ok := m["state_scope"].(map[string]interface{}); ok {
|
||||
sc := &StateScope{}
|
||||
if gs, ok := scope["global"].([]interface{}); ok {
|
||||
for _, g := range gs {
|
||||
if s := asString(g); s != "" {
|
||||
sc.Global = append(sc.Global, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
if ds, ok := scope["depth"].(map[string]interface{}); ok {
|
||||
sc.Depth = map[string]int{}
|
||||
for k, v := range ds {
|
||||
if n, ok := asFloat(v); ok {
|
||||
sc.Depth[k] = int(n)
|
||||
}
|
||||
}
|
||||
sc.Depth = compileDepthMap(sc.Depth)
|
||||
}
|
||||
ui.StateScope = sc
|
||||
}
|
||||
if pg, ok := m["page"].(map[string]interface{}); ok {
|
||||
ui.Page = uipageFromValue(pg)
|
||||
}
|
||||
if pgs, ok := m["pages"].([]interface{}); ok {
|
||||
for _, it := range pgs {
|
||||
if pm, ok := it.(map[string]interface{}); ok {
|
||||
ui.Pages = append(ui.Pages, uipageFromValue(pm))
|
||||
}
|
||||
}
|
||||
}
|
||||
if els, ok := m["elements"].([]interface{}); ok {
|
||||
for _, it := range els {
|
||||
em, ok := it.(map[string]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
el := UIElement{
|
||||
Target: asString(em["target"]),
|
||||
Anchor: asString(em["anchor"]),
|
||||
Mount: asString(em["mount"]),
|
||||
Audience: asString(em["audience"]),
|
||||
}
|
||||
if n, ok := asFloat(em["order"]); ok {
|
||||
el.Order = int(n)
|
||||
}
|
||||
ui.Elements = append(ui.Elements, el)
|
||||
}
|
||||
m, ok := raw.(map[string]interface{})
|
||||
return m, ok
|
||||
}
|
||||
|
||||
// decodeUIExtension reads a plugin's `ui` declaration.
|
||||
//
|
||||
// The field list below is the single source of truth for what the gateway
|
||||
// understands. A declaration using anything else is REPORTED rather than
|
||||
// silently dropped, because the old hand-written reader ignored unknown keys with
|
||||
// no trace: `visibilty = "admin"` (one character off) restricted nothing and
|
||||
// looked exactly like a plugin that simply chose not to declare an audience.
|
||||
func decodeUIExtension(ctx *decodeCtx, m map[string]interface{}) UIExtension {
|
||||
ctx.strictKeys(m, "ui", "visibility", "state_scope", "page", "pages", "elements")
|
||||
|
||||
var ui UIExtension
|
||||
ui.Visibility = decString(m, "visibility")
|
||||
|
||||
if scope := decTable(m, "state_scope"); scope != nil {
|
||||
ctx.strictKeys(scope, "ui.state_scope", "global", "depth")
|
||||
sc := &StateScope{Global: decStringList(scope, "global")}
|
||||
if d := decIntMap(scope, "depth"); len(d) > 0 {
|
||||
sc.Depth = compileDepthMap(d)
|
||||
}
|
||||
ui.StateScope = sc
|
||||
}
|
||||
if ui.Page == nil && len(ui.Pages) == 0 && len(ui.Elements) == 0 {
|
||||
return nil, false
|
||||
|
||||
if pg := decTable(m, "page"); pg != nil {
|
||||
ui.Page = uipageFromValue(pg)
|
||||
}
|
||||
return &ui, true
|
||||
for _, pm := range decTableList(m, "pages") {
|
||||
ui.Pages = append(ui.Pages, uipageFromValue(pm))
|
||||
}
|
||||
for _, em := range decTableList(m, "elements") {
|
||||
ctx.strictKeys(em, "ui.elements[]", "target", "anchor", "mount", "audience", "order")
|
||||
el := UIElement{
|
||||
Target: decString(em, "target"),
|
||||
Anchor: decString(em, "anchor"),
|
||||
Mount: decString(em, "mount"),
|
||||
Audience: decString(em, "audience"),
|
||||
}
|
||||
if n, ok := asFloat(em["order"]); ok {
|
||||
el.Order = int(n)
|
||||
}
|
||||
ui.Elements = append(ui.Elements, el)
|
||||
}
|
||||
return ui
|
||||
}
|
||||
|
||||
func (ps *Plugins) append(p *Plugin) {
|
||||
@ -1140,18 +1159,9 @@ func (ps *Plugins) HookErrors() map[string]map[string]interface{} {
|
||||
// "*" is deliberately NOT a synonym for "all": an audience that looks
|
||||
// restrictive in a config file while meaning everything is exactly the kind of
|
||||
// setting that fails silently.
|
||||
// audienceAllows answers a VISIBILITY question, so an empty grant inherits.
|
||||
func audienceAllows(want, role string) bool {
|
||||
want = strings.ToLower(strings.TrimSpace(want))
|
||||
if want == "" || want == "all" {
|
||||
return true
|
||||
}
|
||||
role = strings.ToLower(strings.TrimSpace(role))
|
||||
for _, part := range strings.Split(want, ",") {
|
||||
if strings.TrimSpace(part) == role {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
return ParseAccessRule(want).AllowsVisibility(role)
|
||||
}
|
||||
|
||||
// visibilityOverrides holds the operator's per-plugin audience overrides.
|
||||
@ -1950,6 +1960,17 @@ func (ps *Plugins) invoke(p *Plugin, fn string, payload map[string]interface{})
|
||||
// quota report is a nested structure with numbers and booleans that must not be
|
||||
// flattened. Hook results still go through invoke, whose map-typed contract the
|
||||
// state merge depends on.
|
||||
// Call invokes an exported function on a plugin's SHARED VM, serialised by the
|
||||
// plugin's mutex.
|
||||
//
|
||||
// The lock is what makes one Lua VM safe to share, and it is held for the whole
|
||||
// call — including any network wait inside it. That is correct for a hook and
|
||||
// wrong for a function that blocks: twenty concurrent quota probes serialise
|
||||
// into the sum of their timeouts (measured: 29s on a 20-source gateway, against
|
||||
// 1.4–3.3s for a single probe with no contention).
|
||||
//
|
||||
// callOn below is the same work without the shared-VM assumption, so a caller
|
||||
// that owns a private VM can use it directly.
|
||||
func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface{}, error) {
|
||||
ps.mu.RLock()
|
||||
var p *Plugin
|
||||
@ -1967,12 +1988,27 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
|
||||
return nil, fmt.Errorf("plugin %s does not export %q", name, fn)
|
||||
}
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
w := p.state
|
||||
if w == nil {
|
||||
p.mu.Unlock()
|
||||
return nil, fmt.Errorf("no state for plugin %s", name)
|
||||
}
|
||||
defer p.mu.Unlock()
|
||||
return callOn(w, fn, args)
|
||||
}
|
||||
|
||||
// callOn runs one exported function on a specific VM.
|
||||
//
|
||||
// It does NOT check the export allowlist and does NOT persist state: both belong
|
||||
// to the shared-VM path in Call. A shard VM has no plugin identity to check
|
||||
// against and no state file of its own — the export allowlist in Call is the
|
||||
// security boundary, and calling this with a function the plugin does not export
|
||||
// is a programming error, not an attack.
|
||||
//
|
||||
// The caller must own w exclusively. No lock is taken here.
|
||||
func callOn(w *worker, fn string, args map[string]interface{}) (interface{}, error) {
|
||||
if w == nil || w.L == nil {
|
||||
return nil, fmt.Errorf("no lua state")
|
||||
}
|
||||
L := w.L
|
||||
L.SetTop(0)
|
||||
defer L.SetTop(0)
|
||||
@ -2005,14 +2041,14 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
|
||||
L.Pop(1)
|
||||
}
|
||||
if errMsg != "" {
|
||||
return nil, fmt.Errorf("plugin %s: %s", name, errMsg)
|
||||
return nil, fmt.Errorf("plugin: %s", errMsg)
|
||||
}
|
||||
if L.GetTop() < 1 || L.IsNil(-1) {
|
||||
return nil, nil
|
||||
}
|
||||
var out interface{}
|
||||
if err := luaToJSON(L, -1, &out); err != nil {
|
||||
return nil, nil
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@ -2097,40 +2133,11 @@ func (h HTTPHandler) routePath() string {
|
||||
//
|
||||
// keyName is the caller key's configured name ("" when it has none), not the
|
||||
// secret.
|
||||
// allows answers a CAPABILITY question, so an empty grant denies. See
|
||||
// AccessRule.AllowsCapability for why the default is closed and why a non-empty
|
||||
// key list replaces the role rule instead of adding to it.
|
||||
func (h HTTPHandler) allows(role, keyName string) bool {
|
||||
// Admin is always admitted. An operator must be able to inspect what a plugin
|
||||
// exposes even while locked out of it, or a typo in a key list would leave
|
||||
// nobody able to diagnose the endpoint they were trying to tighten.
|
||||
if role == "admin" {
|
||||
return true
|
||||
}
|
||||
// Named keys REPLACE the role rule rather than adding to it.
|
||||
if len(h.Keys) > 0 {
|
||||
if keyName == "" {
|
||||
return false
|
||||
}
|
||||
for _, k := range h.Keys {
|
||||
if strings.EqualFold(strings.TrimSpace(k), keyName) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
if h.Public {
|
||||
return true
|
||||
}
|
||||
if len(h.Roles) == 0 {
|
||||
// No declared roles and no public opt-out: treat as admin-only. A plugin
|
||||
// author who forgets this gets a 403 rather than an open endpoint, and
|
||||
// the error says which field to set.
|
||||
return role == "admin"
|
||||
}
|
||||
for _, r := range h.Roles {
|
||||
if strings.EqualFold(strings.TrimSpace(r), role) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
return accessRuleFor(h.Roles, h.Keys, h.Public).AllowsCapability(role, keyName)
|
||||
}
|
||||
|
||||
func (ps *Plugins) StateVisibleTo(name, role string) bool {
|
||||
@ -2176,64 +2183,82 @@ func containsPage(pages []*UIPage, id string) bool {
|
||||
// not exist. Failing the whole plugin would take a working billing widget down
|
||||
// over a typo in an experimental endpoint.
|
||||
func readPluginHTTP(L *golua.State) []HTTPHandler {
|
||||
L.GetGlobal(pluginGlobal)
|
||||
if L.IsNil(-1) {
|
||||
L.SetTop(0)
|
||||
return nil
|
||||
}
|
||||
L.GetField(-1, "http")
|
||||
if L.Type(-1) != golua.LUA_TTABLE {
|
||||
L.SetTop(0)
|
||||
return nil
|
||||
}
|
||||
// luaTableToGoValue, NOT luaToJSON and NOT a hand-rolled GetField loop.
|
||||
// `http` is a LIST of declarations, each naming the handler function by
|
||||
// STRING:
|
||||
//
|
||||
// luaToJSON needs the json global, which is not installed yet while a plugin
|
||||
// is registering. And golua's GetField does NOT convert a numeric key: with
|
||||
// a string index "1" it returns nil, so an index loop silently produced an
|
||||
// empty list and every plugin appeared to declare no endpoints at all.
|
||||
raw := luaTableToGoValue(L, -1)
|
||||
L.SetTop(0)
|
||||
items, ok := raw.([]interface{})
|
||||
// http = {
|
||||
// { path = "health", method = "GET", roles = { "admin" }, handler = "health" },
|
||||
// }
|
||||
//
|
||||
// handler is a name, not a value. An inline function silently produced an
|
||||
// empty handler field and the endpoint was dropped without a word.
|
||||
raw, ok := pluginFieldValue(L, "http")
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
// A single declaration may be given as a bare table instead of a one-element
|
||||
// list; accepting both is friendlier than demanding one spelling.
|
||||
items, isList := raw.([]interface{})
|
||||
if !isList {
|
||||
if one, isTable := raw.(map[string]interface{}); isTable {
|
||||
items = []interface{}{one}
|
||||
} else {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
ctx := &decodeCtx{plugin: "http"}
|
||||
var list []HTTPHandler
|
||||
for _, it := range items {
|
||||
m, ok := it.(map[string]interface{})
|
||||
h, ok := it.(map[string]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
h := HTTPHandler{
|
||||
Path: asString(m["path"]),
|
||||
Method: asString(m["method"]),
|
||||
Handler: asString(m["handler"]),
|
||||
Description: asString(m["description"]),
|
||||
Public: m["public"] == true,
|
||||
ctx.strictKeys(h, "http[]", "path", "method", "handler", "description",
|
||||
"roles", "keys", "public")
|
||||
dh := HTTPHandler{
|
||||
Path: decString(h, "path"),
|
||||
Method: decString(h, "method"),
|
||||
Description: decString(h, "description"),
|
||||
Roles: decStringList(h, "roles"),
|
||||
Keys: decStringList(h, "keys"),
|
||||
}
|
||||
if roles, ok := m["roles"].([]interface{}); ok {
|
||||
for _, r := range roles {
|
||||
if s := asString(r); s != "" {
|
||||
h.Roles = append(h.Roles, s)
|
||||
}
|
||||
// handler must be a NAME, not a value. asString would happily turn a
|
||||
// function into its address ("function: 0x..."), which then failed much
|
||||
// later at dispatch with no connection to the declaration that caused it.
|
||||
// Rejecting it here names the field and the path.
|
||||
if hv, present := h["handler"]; present {
|
||||
name, isStr := hv.(string)
|
||||
if !isStr {
|
||||
ctx.warnf("http[%s]: handler must be the NAME of a function on the "+
|
||||
"plugin table, not the function itself", decString(h, "path"))
|
||||
continue
|
||||
}
|
||||
dh.Handler = name
|
||||
}
|
||||
if keys, ok := m["keys"].([]interface{}); ok {
|
||||
for _, k := range keys {
|
||||
if s := asString(k); s != "" {
|
||||
h.Keys = append(h.Keys, s)
|
||||
}
|
||||
}
|
||||
if p, ok := h["public"].(bool); ok {
|
||||
dh.Public = p
|
||||
}
|
||||
// A malformed entry is dropped rather than fatal: one bad route must not
|
||||
// cost a plugin its hooks and pages. That is also why the mounted count
|
||||
// can be lower than the declared count — /api/plugins lists what mounted.
|
||||
if method, mErr := normalizeHandlerMethod(h.Method); mErr == nil {
|
||||
h.Method = method
|
||||
if validateHandlerPath(h.Path) == nil && h.Handler != "" {
|
||||
list = append(list, h)
|
||||
}
|
||||
// cost a plugin its hooks and pages. The reason is named so a plugin whose
|
||||
// endpoint silently vanished can find out why.
|
||||
method, mErr := normalizeHandlerMethod(dh.Method)
|
||||
switch {
|
||||
case mErr != nil:
|
||||
ctx.warnf("http[%s]: %v", dh.Path, mErr)
|
||||
continue
|
||||
case validateHandlerPath(dh.Path) != nil:
|
||||
ctx.warnf("http[%s]: invalid path", dh.Path)
|
||||
continue
|
||||
case dh.Handler == "":
|
||||
ctx.warnf("http[%s]: handler must name a function on the plugin table "+
|
||||
"(an inline function is not supported)", dh.Path)
|
||||
continue
|
||||
}
|
||||
dh.Method = method
|
||||
list = append(list, dh)
|
||||
}
|
||||
if len(ctx.problems) > 0 {
|
||||
log.Printf("[plugins] %s", strings.Join(ctx.problems, "; "))
|
||||
}
|
||||
return list
|
||||
}
|
||||
|
||||
@ -192,6 +192,11 @@ local function priceFor(payload)
|
||||
out = {
|
||||
prompt = d.prompt or 0, completion = d.completion or 0,
|
||||
per_request = 0, cache_discount = d.cache_discount, peak = d.peak,
|
||||
-- nil = bill reasoning at the completion rate, which is what every
|
||||
-- provider measured actually charges. Set per-rule only for a provider
|
||||
-- that publishes a distinct rate for thinking tokens.
|
||||
reasoning_price = d.reasoning_price,
|
||||
long_context = d.long_context,
|
||||
}
|
||||
|
||||
-- model dimension (a token price overrides the default's token prices)
|
||||
@ -202,6 +207,8 @@ local function priceFor(payload)
|
||||
if mp.completion ~= nil then out.completion = mp.completion end
|
||||
if mp.per_request ~= nil then out.per_request = out.per_request + mp.per_request end
|
||||
if mp.cache_discount ~= nil then out.cache_discount = mp.cache_discount end
|
||||
if mp.reasoning_price ~= nil then out.reasoning_price = mp.reasoning_price end
|
||||
if mp.long_context ~= nil then out.long_context = mp.long_context end
|
||||
if mp.peak ~= nil then out.peak = mp.peak end
|
||||
end
|
||||
|
||||
@ -213,6 +220,8 @@ local function priceFor(payload)
|
||||
if sp.completion ~= nil then out.completion = sp.completion end
|
||||
if sp.per_request ~= nil then out.per_request = out.per_request + sp.per_request end
|
||||
if sp.cache_discount ~= nil then out.cache_discount = sp.cache_discount end
|
||||
if sp.reasoning_price ~= nil then out.reasoning_price = sp.reasoning_price end
|
||||
if sp.long_context ~= nil then out.long_context = sp.long_context end
|
||||
if sp.peak ~= nil then out.peak = sp.peak end
|
||||
end
|
||||
|
||||
@ -346,6 +355,60 @@ local function costFor(payload, price)
|
||||
+ cacheHit * price.prompt * discount
|
||||
+ completion * price.completion * mult
|
||||
|
||||
-- Long-context step: the tokens PAST a threshold cost more.
|
||||
--
|
||||
-- A marginal rate on the excess, not a blended average — that is how
|
||||
-- providers publish it (DeepSeek doubles past 128k) and what keeps it
|
||||
-- composable: a cache read beyond the threshold gets BOTH the cache discount
|
||||
-- AND the long-context rate, as it is billed. A blended average would have to
|
||||
-- choose one, and choosing wrong under-bills exactly the cache-heavy long
|
||||
-- requests the cache exists to make cheap.
|
||||
--
|
||||
-- The excess splits between cached and fresh in the same proportion as the
|
||||
-- whole prompt, so no token is charged twice and none is missed.
|
||||
local lc = price.long_context
|
||||
if lc then
|
||||
local totalPrompt = tonumber(payload.prompt_tokens) or 0
|
||||
local threshold = tonumber(lc.above) or 0
|
||||
local lcRate = tonumber(lc.prompt)
|
||||
if totalPrompt > threshold and threshold > 0 and lcRate and lcRate > 0 then
|
||||
local excess = totalPrompt - threshold
|
||||
-- Only the FRESH share of the excess carries the step: a cache read is
|
||||
-- already billed at the discounted rate, and re-pricing it at the
|
||||
-- long-context rate would charge the same tokens twice.
|
||||
local freshShare = totalPrompt > 0 and (fresh / totalPrompt) or 0
|
||||
local excessFresh = excess * freshShare
|
||||
local excessCached = excess - excessFresh
|
||||
cost = cost
|
||||
+ excessFresh * (lcRate - price.prompt) * mult
|
||||
+ excessCached * (lcRate - price.prompt * discount) * mult
|
||||
end
|
||||
end
|
||||
|
||||
-- Reasoning tokens, when the rule prices them separately.
|
||||
--
|
||||
-- reasoning_tokens is a SUBSET of completion_tokens and is already inside
|
||||
-- the term above. This block only matters when a provider charges thinking
|
||||
-- at a DIFFERENT rate — and every provider measured does not: OpenAI,
|
||||
-- DeepSeek and Anthropic all bill reasoning at the ordinary output rate. So
|
||||
-- reasoning_price is nil by default and this is skipped entirely, which is
|
||||
-- what makes the standard formula match the invoice.
|
||||
--
|
||||
-- When it IS set, only the DIFFERENCE is applied: the reasoning portion is
|
||||
-- re-priced from the normal rate to the reasoning rate. Charging it a
|
||||
-- second time in full would double-bill every thinking token.
|
||||
local reasoningPrice = tonumber(price.reasoning_price)
|
||||
if reasoningPrice and reasoningPrice ~= price.completion then
|
||||
local reasoning = tonumber(payload.reasoning_tokens) or 0
|
||||
if reasoning > 0 then
|
||||
-- Defensive clamp: a provider reporting more reasoning than completion
|
||||
-- would otherwise produce a negative visible-output count and a cost
|
||||
-- delta larger than the whole request.
|
||||
if reasoning > completion then reasoning = completion end
|
||||
cost = cost + reasoning * (reasoningPrice - price.completion) * mult
|
||||
end
|
||||
end
|
||||
|
||||
local flat = price.per_request
|
||||
if not payload.ok and not plugin.count_failures then
|
||||
flat = 0
|
||||
@ -442,6 +505,21 @@ function plugin.on_request_end(payload)
|
||||
s.unpriced_reqs = s.unpriced_reqs + 1
|
||||
local m = payload.model or "?"
|
||||
s.unpriced_models[m] = (s.unpriced_models[m] or 0) + 1
|
||||
|
||||
-- Day-keyed unpriced tally. The lifetime counters above cannot answer the
|
||||
-- question an operator actually has — "is it STILL happening?" — because they
|
||||
-- keep the debt from before a pricing rule was added. After pricing a source
|
||||
-- that was mispriced, the lifetime number never moves, so "unpriced: 6376" on
|
||||
-- a dashboard is indistinguishable from a system that is still billing
|
||||
-- nothing. Counting by day makes the fix observable: the offending day keeps
|
||||
-- its number, new days start at zero.
|
||||
if s.unpriced_by_day == nil then s.unpriced_by_day = {} end
|
||||
-- dayKey wants epoch SECONDS; payload.time is milliseconds.
|
||||
local uday = dayKey(math.floor((tonumber(payload.time) or (os.time() * 1000)) / 1000))
|
||||
local ubucket = s.unpriced_by_day[uday]
|
||||
if ubucket == nil then ubucket = { reqs = 0, models = {} }; s.unpriced_by_day[uday] = ubucket end
|
||||
ubucket.reqs = ubucket.reqs + 1
|
||||
ubucket.models[m] = (ubucket.models[m] or 0) + 1
|
||||
end
|
||||
if s.degraded_reqs == nil then s.degraded_reqs = 0 end
|
||||
-- Degradation is counted here rather than in the chain_step hook because
|
||||
@ -517,7 +595,7 @@ plugin.ui = {
|
||||
global = {
|
||||
"total", "by_source", "by_model", "by_day", "by_day_src",
|
||||
"by_day_model", "by_tier_served", "skip_reasons", "degraded_reqs",
|
||||
"unpriced_models", "unpriced_reqs", "started",
|
||||
"unpriced_models", "unpriced_reqs", "unpriced_by_day", "started",
|
||||
},
|
||||
depth = { by_key = 1, by_day_key = 2 },
|
||||
},
|
||||
@ -593,6 +671,7 @@ plugin.ui = {
|
||||
en: {
|
||||
total: "Total", requests: "Requests", degraded: "Degraded",
|
||||
unpriced: "Unpriced", prompt: "Prompt tokens",
|
||||
unpriced7d: "Unpriced (7d)", unpricedClean: "none in last 7 days",
|
||||
completion: "Completion tokens", failures: "Failures",
|
||||
cacheRate: "Cache hit rate", cacheTokens: "Cache read tokens",
|
||||
perSource: "Per source", perModel: "Per model",
|
||||
@ -614,6 +693,7 @@ plugin.ui = {
|
||||
zh: {
|
||||
total: "总开销", requests: "请求数", degraded: "降级",
|
||||
unpriced: "未定价", prompt: "输入 tokens",
|
||||
unpriced7d: "未定价(近 7 天)", unpricedClean: "近 7 天无",
|
||||
completion: "输出 tokens", failures: "失败",
|
||||
cacheRate: "缓存命中率", cacheTokens: "缓存读取 tokens",
|
||||
perSource: "按源", perModel: "按模型",
|
||||
@ -734,13 +814,35 @@ plugin.ui = {
|
||||
});
|
||||
return out;
|
||||
}
|
||||
// recentUnpriced counts unpriced requests over the last n days. When the
|
||||
// result is zero THAT is the useful signal: every pricing rule added since
|
||||
// then is holding. Null when there is no day data at all (a state file
|
||||
// written before the buckets existed) so the UI can say "unknown" rather than
|
||||
// claim a clean record it cannot support.
|
||||
function recentUnpriced(st, n) {
|
||||
var byDay = st && st.unpriced_by_day;
|
||||
if (!byDay) return null;
|
||||
var cutoff = Date.now() - n * 86400000;
|
||||
var sum = 0, seen = false;
|
||||
Object.keys(byDay).forEach(function (d) {
|
||||
var t = Date.parse(d + "T00:00:00Z");
|
||||
if (isNaN(t)) return;
|
||||
seen = true;
|
||||
if (t >= cutoff) sum += (byDay[d].reqs || 0);
|
||||
});
|
||||
return seen ? sum : null;
|
||||
}
|
||||
|
||||
// rescale returns the state as the selected period presents it. Degraded and
|
||||
// unpriced counters are lifetime-only (there is no day-keyed tally for them)
|
||||
// so they are returned as null and the UI hides them instead of printing a
|
||||
// lifetime number beside a per-window cost — the exact mismatch the period
|
||||
// view exists to remove.
|
||||
function rescale(st) {
|
||||
if (PERIOD === "all" || !st.by_day) return st;
|
||||
if (PERIOD === "all" || !st.by_day) {
|
||||
if (st) st.unpriced_recent = recentUnpriced(st, 7);
|
||||
return st;
|
||||
}
|
||||
var days = daysInWindow(PERIOD, utcToday());
|
||||
var total = zeroBucket();
|
||||
var byDay = {};
|
||||
@ -758,6 +860,10 @@ plugin.ui = {
|
||||
by_day: byDay,
|
||||
degraded_reqs: null,
|
||||
unpriced_reqs: null,
|
||||
// The lifetime unpriced total is null here for the same reason, but the
|
||||
// day buckets are not — they are what lets the card report "0 in the last
|
||||
// 7 days" instead of leaving a settled debt indistinguishable from a leak.
|
||||
unpriced_recent: recentUnpriced(st, 7),
|
||||
};
|
||||
return out;
|
||||
}
|
||||
@ -838,6 +944,15 @@ plugin.ui = {
|
||||
// print a lifetime number next to a per-window cost.
|
||||
view.degraded_reqs != null ? [T.degraded, fmtInt(view.degraded_reqs)] : null,
|
||||
view.unpriced_reqs != null ? [T.unpriced, fmtInt(view.unpriced_reqs)] : null,
|
||||
// "Unpriced (7d)" answers what the lifetime number cannot: is it STILL
|
||||
// happening? A settled debt and a live leak print the same lifetime total.
|
||||
view.unpriced_recent != null
|
||||
? [
|
||||
T.unpriced7d,
|
||||
fmtInt(view.unpriced_recent),
|
||||
view.unpriced_recent === 0 ? T.unpricedClean : null,
|
||||
]
|
||||
: null,
|
||||
[T.prompt, fmtInt(t.prompt_tokens || 0)],
|
||||
[T.completion, fmtInt(t.completion_tokens || 0)],
|
||||
[T.failures, t.failures || 0],
|
||||
@ -949,7 +1064,11 @@ plugin.ui = {
|
||||
monthly: "Monthly fee", peakMult: "Peak multiplier",
|
||||
peakHours: "Peak hours (UTC)", peakDays: "Peak days (UTC 1=Mon)",
|
||||
models: "Per-model prices", addModel: "+ model",
|
||||
mFree: "free", mToken: "token", mSub: "subscription", mUnpriced: "unpriced",
|
||||
|
||||
reasoning: "think $/M",
|
||||
|
||||
lcAbove: "over n tok", lcPrompt: "then $/M", lcMult: "or x",
|
||||
lcHint: "Long-context step: tokens PAST the threshold cost more. Leave all three blank for one flat rate.", reasoningHint: "Thinking tokens are billed as output unless you set a different rate here.", mFree: "free", mToken: "token", mSub: "subscription", mUnpriced: "unpriced",
|
||||
noProfile: "No profile yet — create one to start pricing.",
|
||||
noRules: "No rules in this profile",
|
||||
warnNoSource: "matches no configured source",
|
||||
@ -971,7 +1090,11 @@ plugin.ui = {
|
||||
monthly: "月费", peakMult: "峰段倍数",
|
||||
peakHours: "峰段小时 (UTC)", peakDays: "峰段星期 (UTC 1=周一)",
|
||||
models: "按模型价格", addModel: "+ 模型",
|
||||
mFree: "free 免费", mToken: "token 按量",
|
||||
|
||||
reasoning: "思考 $/M",
|
||||
|
||||
lcAbove: "超过 n tok", lcPrompt: "此后 $/M", lcMult: "或 ×",
|
||||
lcHint: "长上下文阶梯:超过阈值的部分按更高价计。三项都留空 = 单一价格。", reasoningHint: "留空=思考 token 按输出价计(OpenAI/DeepSeek/Anthropic 都这样)。只有供应商对思考单独定价时才填。", mFree: "free 免费", mToken: "token 按量",
|
||||
mSub: "subscription 订阅", mUnpriced: "unpriced 只统计",
|
||||
noProfile: "还没有方案 —— 新建一个开始定价。",
|
||||
noRules: "该方案暂无规则",
|
||||
@ -1069,6 +1192,25 @@ plugin.ui = {
|
||||
"<input class='m-name' value='" + esc(m) + "' style='flex:1 1 90px;min-width:0'>" +
|
||||
"<input class='m-p' value='" + esc(t.prompt == null ? "" : t.prompt) +
|
||||
"' placeholder='" + esc(T.prompt) + "' style='flex:0 1 62px;min-width:0'>" +
|
||||
// Left BLANK by default on purpose: empty means "bill thinking at the
|
||||
// completion rate", which is what OpenAI, DeepSeek and Anthropic do.
|
||||
// A 0 here would mean "thinking is free" and would under-bill quietly.
|
||||
"<input class='m-r' value='" + esc(t.reasoning == null ? "" : t.reasoning) +
|
||||
"' placeholder='" + esc(T.reasoning) + "' title='" + esc(T.reasoningHint) +
|
||||
"' style='flex:0 1 62px;min-width:0'>" +
|
||||
// Long-context step: threshold + (absolute rate OR multiplier). All
|
||||
// three blank means NO step, deliberately distinct from a step whose
|
||||
// rate is 0 — three inputs rather than inferring the step from a zero.
|
||||
// On an agent workload 88% of requests exceed 128k prompt tokens, so
|
||||
// omitting this under-bills the MAJORITY while every number stays
|
||||
// plausible. The step is MARGINAL: it reprices the excess only.
|
||||
"<input class='m-lc-a' value='" + esc(t.long_context == null ? "" : t.long_context.above) +
|
||||
"' placeholder='" + esc(T.lcAbove) + "' title='" + esc(T.lcHint) +
|
||||
"' style='flex:0 1 52px;min-width:0'>" +
|
||||
"<input class='m-lc-p' value='" + esc(t.long_context == null ? "" : (t.long_context.prompt || "")) +
|
||||
"' placeholder='" + esc(T.lcPrompt) + "' style='flex:0 1 52px;min-width:0'>" +
|
||||
"<input class='m-lc-m' value='" + esc(t.long_context == null ? "" : (t.long_context.multiplier || "")) +
|
||||
"' placeholder='" + esc(T.lcMult) + "' style='flex:0 1 46px;min-width:0'>" +
|
||||
"<input class='m-c' value='" + esc(t.completion == null ? "" : t.completion) +
|
||||
"' placeholder='" + esc(T.completion) + "' style='flex:0 1 62px;min-width:0'>" +
|
||||
"<button class='ghost small m-del' title='" + esc(T.delRule) + "' style='flex:0 0 auto'>×</button></div>";
|
||||
@ -1145,6 +1287,39 @@ plugin.ui = {
|
||||
return;
|
||||
}
|
||||
models[n] = { prompt: pv, completion: cv };
|
||||
|
||||
// Optional thinking-token rate, read from the same row. Blank means
|
||||
// "bill reasoning at the completion rate" — what OpenAI, DeepSeek and
|
||||
// Anthropic all do. A 0 would mean "thinking is free", so an empty
|
||||
// field must stay ABSENT from the saved rule rather than becoming 0.
|
||||
var rel = r.querySelector(".m-r");
|
||||
var rtxt = rel ? String(rel.value || "").trim() : "";
|
||||
if (rtxt !== "") {
|
||||
if (isNaN(parseFloat(rtxt))) {
|
||||
bad = n + " (" + T.reasoning + ")";
|
||||
return;
|
||||
}
|
||||
models[n].reasoning = rtxt;
|
||||
}
|
||||
// Long-context step. All three blank means NO step — the key must be
|
||||
// ABSENT from the saved rule rather than present-and-zero, which would
|
||||
// be a step priced at nothing.
|
||||
var lca = r.querySelector(".m-lc-a");
|
||||
var lcp = r.querySelector(".m-lc-p");
|
||||
var lcm = r.querySelector(".m-lc-m");
|
||||
var lcaT = lca ? String(lca.value || "").trim() : "";
|
||||
var lcpT = lcp ? String(lcp.value || "").trim() : "";
|
||||
var lcmT = lcm ? String(lcm.value || "").trim() : "";
|
||||
if (lcaT !== "" || lcpT !== "" || lcmT !== "") {
|
||||
if (lcaT === "" || isNaN(parseFloat(lcaT))) { bad = n + " (" + T.lcAbove + ")"; return; }
|
||||
if (lcpT !== "" && isNaN(parseFloat(lcpT))) { bad = n + " (" + T.lcPrompt + ")"; return; }
|
||||
if (lcmT !== "" && isNaN(parseFloat(lcmT))) { bad = n + " (" + T.lcMult + ")"; return; }
|
||||
if (lcpT === "" && lcmT === "") { bad = n + " (" + T.lcPrompt + "/" + T.lcMult + ")"; return; }
|
||||
var lc = { above: String(parseFloat(lcaT)) };
|
||||
if (lcpT !== "") lc.prompt = lcpT;
|
||||
if (lcmT !== "") lc.multiplier = parseFloat(lcmT);
|
||||
models[n].long_context = lc;
|
||||
}
|
||||
});
|
||||
if (bad) throw new Error(T.needPrices + ": " + bad);
|
||||
if (Object.keys(models).length) out.models = models;
|
||||
@ -1335,6 +1510,11 @@ plugin.ui = {
|
||||
"<input class='m-name' placeholder='model' style='flex:1;min-width:80px'>" +
|
||||
"<input class='m-p' placeholder='" + esc(T.prompt) + "' style='width:96px'>" +
|
||||
"<input class='m-c' placeholder='" + esc(T.completion) + "' style='width:96px'>" +
|
||||
"<input class='m-r' placeholder='" + esc(T.reasoning) + "' title='" + esc(T.reasoningHint) +
|
||||
"' style='width:96px'>" +
|
||||
"<input class='m-lc-a' placeholder='" + esc(T.lcAbove) + "' title='" + esc(T.lcHint) + "' style='width:78px'>" +
|
||||
"<input class='m-lc-p' placeholder='" + esc(T.lcPrompt) + "' style='width:78px'>" +
|
||||
"<input class='m-lc-m' placeholder='" + esc(T.lcMult) + "' style='width:70px'>" +
|
||||
"<button class='ghost small m-del'>×</button>";
|
||||
box.appendChild(row);
|
||||
};
|
||||
|
||||
@ -43,7 +43,7 @@ PLUGIN.name = "quota"
|
||||
PLUGIN.version = "1.0.0"
|
||||
PLUGIN.description = "多源套餐余量(余额 / 用量 / 滚动窗口)"
|
||||
|
||||
PLUGIN.exports = { "collect_source", "collect_all" }
|
||||
PLUGIN.exports = { "collect_source", "collect_all", "due_sources", "memo_record" }
|
||||
|
||||
-- plugin.state must exist before anything indexes it. The kernel restores
|
||||
-- persisted state OVER this table, so declaring it here is safe: a restart with
|
||||
@ -63,8 +63,23 @@ end
|
||||
-- hostOf extracts the host from a base_url, for showing WHICH upstream a row is
|
||||
-- about. Two sources on the same vendor (two Command Code keys) are
|
||||
-- indistinguishable without it.
|
||||
-- hostOf returns the host of a URL — the part memoisation is keyed on, so two
|
||||
-- sources pointing at the same host with different paths share one answer.
|
||||
--
|
||||
-- It used to be written with string.gmatch and a capture group. gmatch yields
|
||||
-- WHOLE matches, not captures, and with `^` it never advances, so the iterator
|
||||
-- returned nil on the first step and hostOf returned its `or u` fallback —
|
||||
-- the FULL URL.
|
||||
--
|
||||
-- The consequence was invisible and total. Every lookup was HOST_MEMO[nil] and
|
||||
-- every write was HOST_MEMO[nil], so:
|
||||
-- - the memo table never returned anything, for any host, ever
|
||||
-- - the negative memo, the TTL and (later) the backoff were all dead code
|
||||
-- - the 19 entries in this gateway's production state are unreachable: their
|
||||
-- keys are full URLs, while a lookup asks for nil
|
||||
-- Measured before the fix: hostOf(key) matched 0 of 19 production entries.
|
||||
local function hostOf(u)
|
||||
return (string.gmatch(u, "^https?://([^/]+)")()) or u
|
||||
return (string.match(u, "^https?://([^/]+)")) or u
|
||||
end
|
||||
|
||||
-- get walks a dotted path through nested tables, returning nil if any hop is
|
||||
@ -298,6 +313,52 @@ local HOST_MEMO = {}
|
||||
local HOST_MEMO_AT = {}
|
||||
local PROBE_MEMO_TTL = 1800 -- 30 minutes
|
||||
|
||||
-- Exponential backoff for hosts that answer nothing.
|
||||
--
|
||||
-- A flat TTL is the wrong shape for a dead upstream. Measured on this gateway:
|
||||
-- two hosts had been unreachable for 10 HOURS, and because PROBE_MEMO_TTL is 30
|
||||
-- minutes their negative memo was stale on every single page open — so each
|
||||
-- refresh re-paid the full cost of three adapters (measured: 24s per source
|
||||
-- before the probe budget was fixed, 16s after). A host that is down should not
|
||||
-- be re-examined every 30 minutes forever; each consecutive failure doubles the
|
||||
-- wait, so a long-dead host costs almost nothing while a host that was briefly
|
||||
-- down is retried within the TTL as before.
|
||||
--
|
||||
-- The cap matters as much as the doubling: without it a host that is down for a
|
||||
-- week stops being probed at all, and the page would quietly and permanently
|
||||
-- misreport it as having no balance endpoint.
|
||||
--
|
||||
-- SUCCESS resets the streak. This is only a probe-scheduling decision — a source
|
||||
-- that stops answering is still shown, still counted, and still retried.
|
||||
local BACKOFF_BASE = 1800 -- first retry after the usual TTL
|
||||
local BACKOFF_FACTOR = 2 -- doubles per consecutive miss
|
||||
local BACKOFF_MAX = 21600 -- 6 hours; beyond this we always re-probe
|
||||
|
||||
-- backoffSeconds returns how long to wait before re-probing a host that has
|
||||
-- missed `streak` times in a row, and 0 when it is due now.
|
||||
--
|
||||
-- `at` is when the host last failed. streak is stored inside the memo table
|
||||
-- itself so it survives a restart: a process restart must not reset the backoff
|
||||
-- and re-pay the cost it was avoiding.
|
||||
-- Returns a BOOLEAN and nothing else.
|
||||
--
|
||||
-- In Lua only nil and false are falsy, so every path here must return one of
|
||||
-- those. Two versions of this got it wrong and both were caught by a test that
|
||||
-- asserts the type: `return 0` for "no streak" (0 is truthy, so a host with no
|
||||
-- streak — which is every memo written before this change — would be skipped
|
||||
-- forever, silently), and `return BACKOFF_MAX` on the cap path (21600 is
|
||||
-- truthy too, so a long-dead host would never be re-probed, which is the exact
|
||||
-- failure the cap exists to prevent).
|
||||
local function backoffSeconds(at, streak)
|
||||
if streak == nil or streak <= 0 then return false end
|
||||
local wait = BACKOFF_BASE
|
||||
for _ = 2, streak do
|
||||
wait = wait * BACKOFF_FACTOR
|
||||
if wait >= BACKOFF_MAX then wait = BACKOFF_MAX break end
|
||||
end
|
||||
return (os.time() - (at or 0)) < wait
|
||||
end
|
||||
|
||||
-- PROBE_BUDGET_MS caps one collect pass. Without it a page open over a set of
|
||||
-- dead upstreams runs to the sum of every timeout — measured at 149s — and the
|
||||
-- plugin's write lock is held the whole time.
|
||||
@ -325,6 +386,12 @@ local function collectSource(src, budget)
|
||||
-- exactly the confusion that made an earlier revision drop five sources.
|
||||
if type(known) == "table" then
|
||||
local seen = known.at or 0
|
||||
-- A streak is a host that has failed repeatedly. Skip it outright until its
|
||||
-- backoff expires, instead of re-paying three adapter timeouts every TTL.
|
||||
if backoffSeconds(seen, known.streak) then
|
||||
return { name = src.name, host = host, skipped = true,
|
||||
reason = "backing off after " .. tostring(known.streak) .. " failed probes" }
|
||||
end
|
||||
if PROBE_MEMO_TTL > 0 and (os.time() - seen) > PROBE_MEMO_TTL then
|
||||
known = nil
|
||||
else
|
||||
@ -394,8 +461,21 @@ local function collectSource(src, budget)
|
||||
-- would make the page quietly stop reporting a source that started working.
|
||||
-- The stamp is what turns a 60s first load into a fast steady state without
|
||||
-- freezing what the page can see.
|
||||
if rep == nil and not HOST_MEMO[host] then
|
||||
HOST_MEMO[host] = { at = os.time() }
|
||||
-- Record the negative result, and GROW THE STREAK.
|
||||
--
|
||||
-- The guard used to be `not HOST_MEMO[host]`, which is true only the FIRST time
|
||||
-- a host fails: on every later failure the memo already held a table, the
|
||||
-- guard was false, and nothing was written. The streak therefore stayed pinned
|
||||
-- at 1 forever and the backoff never grew. Measured while writing this: the
|
||||
-- memo read back `{at:... streak:1}` after four consecutive failures.
|
||||
--
|
||||
-- A success overwrites the memo with a vendor string, so a host that recovers
|
||||
-- resets its streak automatically.
|
||||
if rep == nil then
|
||||
local prev = HOST_MEMO[host]
|
||||
local streak = 1
|
||||
if type(prev) == "table" then streak = (prev.streak or 1) + 1 end
|
||||
HOST_MEMO[host] = { at = os.time(), streak = streak }
|
||||
end
|
||||
if rep ~= nil then
|
||||
HOST_MEMO_AT[host] = os.time()
|
||||
@ -421,6 +501,92 @@ end
|
||||
-- and doing 21 sources in sequence overran the HTTP call's own budget — the
|
||||
-- whole page returned 502 while individual probes were still in flight. The
|
||||
-- sequential entry point stays for tests and for anyone scripting the plugin.
|
||||
-- due_sources answers, on the SHARED VM, which sources are worth probing right
|
||||
-- now — and it makes NO network requests.
|
||||
--
|
||||
-- This exists because of a conflict between two facts:
|
||||
--
|
||||
-- - Probing is network-bound, and Plugins.Call holds the plugin mutex for the
|
||||
-- whole Lua function, so twenty probes through /call/ serialise. Measured:
|
||||
-- 29s on a 20-source gateway. The fix is ShardCall, which runs each call on
|
||||
-- its own VM — and a shard VM has its own PLUGIN.state.
|
||||
-- - The backoff lives in PLUGIN.state (HOST_MEMO). A shard therefore cannot
|
||||
-- see it, cannot record it, and a host that fails keeps being probed on every
|
||||
-- page open. Measured: fanout left the memo at 0 entries, collect_all reached
|
||||
-- 17.
|
||||
--
|
||||
-- So the split is: the DECISION runs where the state is (one shared-VM call,
|
||||
-- microseconds), and only the hosts that pass it get sharded.
|
||||
--
|
||||
-- Cheap on purpose. It reads HOST_MEMO and nothing else; if the state has not
|
||||
-- been restored yet, every host is due, which is the correct answer for a first
|
||||
-- call.
|
||||
-- memo_record writes the outcome of a sharded pass into the SHARED VM's memo.
|
||||
--
|
||||
-- A shard runs on its own VM and therefore cannot record anything: measured, a
|
||||
-- page of fanouts left the memo at 0 entries, so the backoff never accumulated
|
||||
-- and every refresh re-probed the dead hosts. This is the other half of
|
||||
-- due_sources — the decision is taken here, the outcome is written back here, and
|
||||
-- only the network wait happens on the shards.
|
||||
--
|
||||
-- It records FAILURES as well as successes. Recording only successes is what
|
||||
-- left every source with no balance endpoint re-probing three adapters forever.
|
||||
PLUGIN.memo_record = function(payload)
|
||||
if type(payload) ~= "table" then return nil, "expected table" end
|
||||
if not next(HOST_MEMO) then
|
||||
local st = PLUGIN.state
|
||||
if type(st) == "table" and type(st.probe_memo) == "table" then
|
||||
for k, v in pairs(st.probe_memo) do HOST_MEMO[k] = v end
|
||||
end
|
||||
end
|
||||
local rows = payload.results
|
||||
if type(rows) ~= "table" then return nil, "missing results" end
|
||||
local now = os.time()
|
||||
for _, r in ipairs(rows) do
|
||||
if type(r) == "table" and type(r.host) == "string" and r.host ~= "" then
|
||||
local host = hostOf(r.host)
|
||||
if r.vendor then
|
||||
-- A source that answered resets its streak: the memo becomes a vendor
|
||||
-- string, which every reader already treats as a positive memo.
|
||||
HOST_MEMO[host] = r.vendor
|
||||
else
|
||||
local prev = HOST_MEMO[host]
|
||||
local streak = 1
|
||||
if type(prev) == "table" then streak = (prev.streak or 1) + 1 end
|
||||
HOST_MEMO[host] = { at = now, streak = streak }
|
||||
end
|
||||
end
|
||||
end
|
||||
PLUGIN.state.probe_memo = HOST_MEMO
|
||||
return true
|
||||
end
|
||||
|
||||
PLUGIN.due_sources = function(payload)
|
||||
if type(payload) ~= "table" then return nil, "expected table" end
|
||||
if not next(HOST_MEMO) then
|
||||
local st = PLUGIN.state
|
||||
if type(st) == "table" and type(st.probe_memo) == "table" then
|
||||
for k, v in pairs(st.probe_memo) do HOST_MEMO[k] = v end
|
||||
end
|
||||
end
|
||||
local srcs = payload.sources
|
||||
if type(srcs) ~= "table" then return nil, "missing sources" end
|
||||
local due, backing = {}, {}
|
||||
for _, s in ipairs(srcs) do
|
||||
if type(s) == "table" and type(s.base_url) == "string" and s.base_url ~= "" then
|
||||
local base = (s.base_url:gsub("/+$", ""))
|
||||
local known = HOST_MEMO[hostOf(base)]
|
||||
if type(known) == "table" and backoffSeconds(known.at or 0, known.streak) then
|
||||
backing[#backing + 1] = { name = s.name, host = hostOf(base),
|
||||
reason = "backing off after " .. tostring(known.streak) .. " failed probes" }
|
||||
else
|
||||
due[#due + 1] = s
|
||||
end
|
||||
end
|
||||
end
|
||||
return { due = due, backing_off = backing }
|
||||
end
|
||||
|
||||
PLUGIN.collect_all = function(payload)
|
||||
if type(payload) ~= "table" then return nil, "expected table" end
|
||||
if not next(HOST_MEMO) then
|
||||
@ -431,10 +597,15 @@ PLUGIN.collect_all = function(payload)
|
||||
end
|
||||
local srcs = payload.sources
|
||||
if type(srcs) ~= "table" then return nil, "missing sources" end
|
||||
local t1 = os.clock()
|
||||
-- os.time, NOT os.clock. os.clock measures CPU time, and a probe spends
|
||||
-- essentially none: measured on this gateway, 3s blocked on an HTTP request
|
||||
-- advanced os.clock by 0.001s. A budget written in os.clock therefore never
|
||||
-- fires, which is why a dead upstream ran the full
|
||||
-- 4s timeout x 2 retries x 3 adapters = 24s against a "12 second" budget.
|
||||
local t1 = os.time()
|
||||
local lim1 = PROBE_BUDGET_MS / 1000
|
||||
local function spent()
|
||||
return (os.clock() - t1) > lim1
|
||||
return (os.time() - t1) > lim1
|
||||
end
|
||||
local budget = { exhausted = spent }
|
||||
local out = {}
|
||||
@ -463,10 +634,10 @@ PLUGIN.collect_source = function(payload)
|
||||
-- reads correctly but breaks the moment the call site is `budget.exhausted()`
|
||||
-- from inside a pcall-guarded hook where the upvalue chain differs; the
|
||||
-- explicit capture below cannot go wrong that way.
|
||||
local t0 = os.clock()
|
||||
local t0 = os.time()
|
||||
local limit = PROBE_BUDGET_MS / 1000
|
||||
local function exhausted()
|
||||
return (os.clock() - t0) > limit
|
||||
return (os.time() - t0) > limit
|
||||
end
|
||||
local budget = { exhausted = exhausted }
|
||||
for _, s in ipairs(srcs) do
|
||||
@ -732,15 +903,35 @@ PLUGIN.ui = {
|
||||
var sub=document.getElementById("qa-sub"); if(sub) sub.textContent="探测中…";
|
||||
var out;
|
||||
try{
|
||||
// One call per source, fired together. Probing is network-bound, so a
|
||||
// sequential collect_all over 21 sources blows past the call's time budget
|
||||
// and every card comes back as one 502 — the probes themselves were fine.
|
||||
// Concurrency also makes the page honest about which sources are slow:
|
||||
// each card appears as its own answer lands.
|
||||
out = await Promise.all(srcs.map(function(s){
|
||||
return pluginAPI.call("quota","collect_source",{sources:[s],name:s.name})
|
||||
.catch(function(e){ return {name:s.name,skipped:true,error:String(e.message||e)}; });
|
||||
}));
|
||||
// ONE request, fanned out inside the gateway.
|
||||
//
|
||||
// The previous version fired Promise.all at /call/ — twenty concurrent
|
||||
// browser requests. That looked concurrent and was not: each one queued on
|
||||
// the plugin mutex for the whole Lua call, network wait included. Measured on
|
||||
// the live gateway: 29s of wall clock for twenty probes, against 1.4–3.3s for
|
||||
// a single probe with no contention. The page sat on "探测中…" for half a
|
||||
// minute and read as broken.
|
||||
//
|
||||
// The comment that used to sit here claimed concurrency avoided a call budget
|
||||
// overrun. It could not have: the requests were serialised, so the budget they
|
||||
// overran was the very same one. It is a single call now because a
|
||||
// request-per-item fan-out cannot be parallel from the client at all.
|
||||
// Which sources are worth probing is decided on the SHARED VM, where the
|
||||
// backoff state lives; only those get sharded.
|
||||
//
|
||||
// Asking each shard to decide for itself does not work: a shard has its own
|
||||
// PLUGIN.state, so it cannot see the memo and a host that keeps failing is
|
||||
// re-probed on every refresh forever. Measured before this split: fanout left
|
||||
// the memo at 0 entries and every refresh cost the full 12s.
|
||||
var plan = await pluginAPI.call("quota","due_sources",{sources:srcs});
|
||||
var due = (plan && plan.due) || [];
|
||||
var out = (plan && plan.backing_off) || [];
|
||||
|
||||
if (due.length) {
|
||||
out = out.concat(await pluginAPI.fanout("quota","collect_source",due.map(function(s){
|
||||
return {sources:[s],name:s.name};
|
||||
})));
|
||||
}
|
||||
// Mirror the host→vendor map the plugin keeps: a source that answered is
|
||||
// worth probing first next time, and the page has everything it needs to
|
||||
// record that without another round trip.
|
||||
@ -749,6 +940,18 @@ PLUGIN.ui = {
|
||||
if(r && !r.skipped && !r.error && r.host && r.vendor) memo[r.host] = r.vendor;
|
||||
});
|
||||
window.__qaMemo = memo;
|
||||
// Hand the whole pass back to the SHARED VM so the backoff has something to
|
||||
// work from next time. Without this the shards' results die with their VMs:
|
||||
// measured, the memo stayed at 0 entries and every refresh paid full price.
|
||||
await pluginAPI.call("quota","memo_record",{results:(out||[]).map(function(r){
|
||||
if(!r) return null;
|
||||
// A host that is merely BACKING OFF has not just failed; recording it as a
|
||||
// failure would grow the streak on a probe that never happened.
|
||||
if(r.reason && String(r.reason).indexOf("backing off")===0) return null;
|
||||
if(r.error) return {host:r.host};
|
||||
if(r.skipped) return {host:r.host};
|
||||
return {host:r.host, vendor:r.vendor};
|
||||
}).filter(Boolean)});
|
||||
}catch(e){
|
||||
if(btn) btn.disabled=false;
|
||||
if(sub) sub.textContent="探测失败: "+e.message;
|
||||
|
||||
556
internal/lua/quota_backoff_test.go
Normal file
556
internal/lua/quota_backoff_test.go
Normal file
@ -0,0 +1,556 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The backoff lives in quota.lua — a shipped plugin, not Go — so these tests read
|
||||
// the file that actually ships (readShippedPlugin) rather than a copy that can
|
||||
// drift from it.
|
||||
|
||||
// TestQuotaBackoffIsAppliedInSource guards the wiring.
|
||||
//
|
||||
// It exists because of a mutation that removed the call site entirely (`if false
|
||||
// then`) and every other test in this file still passed. The other tests cover
|
||||
// the decision FUNCTION and the streak RECORDING; neither exists in order to be
|
||||
// unused. A backoff nobody consults is indistinguishable from no backoff, and the
|
||||
// symptom is the original one — a dead host re-probed on every page open.
|
||||
func TestQuotaBackoffIsAppliedInSource(t *testing.T) {
|
||||
src := readShippedPlugin(t, "quota.lua")
|
||||
if !strings.Contains(src, "if backoffSeconds(seen, known.streak) then") {
|
||||
t.Error("quota.lua never consults backoffSeconds for a host with a streak — " +
|
||||
"the backoff is computed but not applied")
|
||||
}
|
||||
if !strings.Contains(src, `reason = "backing off after "`) {
|
||||
t.Error("the skip path does not say why it skipped; a silently skipped host " +
|
||||
"looks exactly like a host that simply has no balance endpoint")
|
||||
}
|
||||
// The recording guard has to REPLACE the memo, not merge into it. The bug this
|
||||
// guards was `if rep == nil and not HOST_MEMO[host]`, which is true only on a
|
||||
// host's first failure — so the streak stayed pinned at 1 forever. Asserted on
|
||||
// the source because the mutation that reintroduces it leaves the visible
|
||||
// behaviour of the FIRST collect unchanged, and that is all the integration
|
||||
// test can reach inside one process.
|
||||
if !strings.Contains(src, "if rep == nil then") {
|
||||
t.Error("the negative result is recorded under a guard that skips every " +
|
||||
"failure after the first — the streak will never grow")
|
||||
}
|
||||
if strings.Contains(src, "and not HOST_MEMO[host] then") {
|
||||
t.Error("the recording guard excludes a host that already has a memo, so " +
|
||||
"consecutive failures are never counted")
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaBackoffWindowIsCapped guards the failure mode an exponential backoff
|
||||
// has when it is not capped: a host that is down for a week stops being probed
|
||||
// and the page permanently misreports it as having no balance endpoint.
|
||||
func TestQuotaBackoffWindowIsCapped(t *testing.T) {
|
||||
src := readShippedPlugin(t, "quota.lua")
|
||||
for _, want := range []string{"BACKOFF_BASE", "BACKOFF_FACTOR", "BACKOFF_MAX"} {
|
||||
if !strings.Contains(src, want) {
|
||||
t.Errorf("quota.lua no longer declares %s — the backoff was weakened or removed", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaBackoffWindows walks the whole policy through the plugin's own
|
||||
// decision function: for each consecutive failure, how long the host must be left
|
||||
// alone, that it is still skipped halfway through that window, that it is retried
|
||||
// once the window passes, and that an absurdly old host is always due.
|
||||
//
|
||||
// Why the function rather than the running plugin: HOST_MEMO is a process-local
|
||||
// cache, restored on a process's first collect and written back afterwards, so
|
||||
// ageing its memo from Go changes nothing the plugin can see. Two earlier versions
|
||||
// of this file spent a long time on exactly that dead end, and a test that keeps
|
||||
// breaking for reasons unrelated to the code under test is worse than no test.
|
||||
//
|
||||
// The function is the plugin's own — reached through a shim that appends one
|
||||
// export — so changing quota.lua's arithmetic moves this test. It is not a
|
||||
// reimplementation that would agree with a broken plugin.
|
||||
func TestQuotaBackoffWindows(t *testing.T) {
|
||||
ps := quotaWithDecision(t)
|
||||
|
||||
// streak, wait seconds. The last rows are at and past the ceiling.
|
||||
for _, c := range []struct {
|
||||
streak int
|
||||
wait int
|
||||
}{
|
||||
{1, 1800},
|
||||
{2, 3600},
|
||||
{3, 7200},
|
||||
{4, 14400},
|
||||
{5, 21600},
|
||||
{9, 21600},
|
||||
{20, 21600},
|
||||
} {
|
||||
if got := quotaSkip(t, ps, c.streak, c.wait/2); !got {
|
||||
t.Logf(" streak=%d age=%d → %v", c.streak, c.wait/2, got)
|
||||
t.Errorf("streak=%d: not skipping %ds into a %ds window — the wait is too short",
|
||||
c.streak, c.wait/2, c.wait)
|
||||
}
|
||||
if quotaSkip(t, ps, c.streak, c.wait+60) {
|
||||
t.Errorf("streak=%d: still skipping %ds after its %ds window expired — "+
|
||||
"this host would never be retried again", c.streak, c.wait+60, c.wait)
|
||||
}
|
||||
if quotaSkip(t, ps, c.streak, 1<<30) {
|
||||
t.Errorf("streak=%d: a host silent for decades is still being skipped", c.streak)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaBackoffReturnsOnlyBooleans pins a type rule that produced two silent
|
||||
// bugs.
|
||||
//
|
||||
// In Lua only nil and false are falsy, so every path of a predicate must return
|
||||
// one of those. This function had two paths returning numbers: `return 0` for
|
||||
// "no streak", which is truthy and made every caller skip a host that had never
|
||||
// been probed; and `return BACKOFF_MAX` on the cap path, also truthy, which made
|
||||
// the ceiling mean "never retried" — the exact failure the ceiling exists to
|
||||
// prevent.
|
||||
//
|
||||
// Both shipped far enough to be found by this assertion rather than by a user
|
||||
// staring at a page that had quietly stopped probing anything.
|
||||
func TestQuotaBackoffReturnsOnlyBooleans(t *testing.T) {
|
||||
ps := quotaWithDecision(t)
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
age int
|
||||
strk interface{}
|
||||
}{
|
||||
{"no streak field (every memo written before the backoff existed)", 0, nil},
|
||||
{"zero streak", 0, 0},
|
||||
{"negative streak", 0, -1},
|
||||
{"inside the window", 10, 1},
|
||||
{"outside the window", 999999, 1},
|
||||
{"at the ceiling", 10, 20},
|
||||
} {
|
||||
out, err := ps.Call("quota", "__decision", map[string]interface{}{
|
||||
"at": float64(time.Now().Unix() - int64(c.age)),
|
||||
"streak": c.strk,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", c.name, err)
|
||||
}
|
||||
// A bare boolean is a valid answer: when backoffSeconds returns false the
|
||||
// shim's table round-trips as just `false`, because a Lua table holding a
|
||||
// single falsy field collapses on encode.
|
||||
switch v := out.(type) {
|
||||
case bool:
|
||||
_ = v // a boolean is exactly what this asserts
|
||||
case map[string]interface{}:
|
||||
if _, ok := v["r"].(bool); !ok {
|
||||
t.Errorf("%s: backoffSeconds returned %T (%v) — in Lua a number is "+
|
||||
"TRUTHY, so this caller would skip the host", c.name, v["r"], v["r"])
|
||||
}
|
||||
default:
|
||||
t.Errorf("%s: decision returned %T (%v); want a boolean", c.name, out, out)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaBackoffRecordsConsecutiveFailures covers the recording guard through
|
||||
// the running plugin, because that part IS observable from outside.
|
||||
//
|
||||
// The guard used to be `not HOST_MEMO[host]`, true only on a host's first
|
||||
// failure. After that the memo held a table, the guard was false, nothing was
|
||||
// written, and the streak stayed pinned at 1 forever — measured by reading back
|
||||
// {at:... streak:1} after four consecutive failures.
|
||||
//
|
||||
// Only two things are asserted, because only two are reachable in one process:
|
||||
// a first failure records streak 1, and a SKIP does not grow it. A skipped probe
|
||||
// is not a failure, and a skip that counted would let a never-probed host grow
|
||||
// without end and never be retried.
|
||||
func TestQuotaBackoffRecordsConsecutiveFailures(t *testing.T) {
|
||||
srv := newCountingServer(t)
|
||||
ps := newQuotaPlugins(t, srv.URL)
|
||||
|
||||
args := map[string]interface{}{
|
||||
"sources": []interface{}{map[string]interface{}{
|
||||
"name": "dead", "base_url": srv.URL + "/v1", "api_key": "k",
|
||||
}},
|
||||
"name": "dead",
|
||||
}
|
||||
streak := func() float64 {
|
||||
st, _ := ps.State("quota").(map[string]interface{})
|
||||
memo, _ := st["probe_memo"].(map[string]interface{})
|
||||
for _, v := range memo {
|
||||
if e, ok := v.(map[string]interface{}); ok {
|
||||
if f, ok := e["streak"].(float64); ok {
|
||||
return f
|
||||
}
|
||||
}
|
||||
}
|
||||
t.Fatalf("no negative memo recorded: %#v", memo)
|
||||
return 0
|
||||
}
|
||||
|
||||
if _, err := ps.Call("quota", "collect_source", args); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := streak(); got != 1 {
|
||||
t.Fatalf("streak after the first failure = %.0f, want 1", got)
|
||||
}
|
||||
|
||||
before := srv.Hits()
|
||||
if _, err := ps.Call("quota", "collect_source", args); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if served := srv.Hits() - before; served != 0 {
|
||||
t.Errorf("a host inside its backoff window still made %d requests — the "+
|
||||
"whole feature is to not make any", served)
|
||||
}
|
||||
if got := streak(); got != 1 {
|
||||
t.Errorf("a SKIP changed the streak to %.0f — a skipped probe is not a failure", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaMemoKeyIsTheHost pins the key the memo is written and read under.
|
||||
//
|
||||
// This is the foundation the whole memoisation rests on, and it was broken in the
|
||||
// most total way possible: hostOf was written with string.gmatch and a capture
|
||||
// group, gmatch yields whole matches rather than captures, and with `^` the
|
||||
// iterator never advanced — so it returned nil and every lookup and every write
|
||||
// went to HOST_MEMO[nil]. The memo therefore never returned anything for any
|
||||
// host, and the negative memo, the TTL and the backoff on top of them were all
|
||||
// dead code that looked alive.
|
||||
//
|
||||
// Measured on this gateway before the fix: hostOf(key) matched 0 of the 19
|
||||
// entries in production state.
|
||||
//
|
||||
// The check is on the shipped source plus its runtime behaviour, because the
|
||||
// failure mode is invisible from outside — a memo that never matches anything
|
||||
// behaves exactly like a memo that has nothing to remember.
|
||||
func TestQuotaMemoKeyIsTheHost(t *testing.T) {
|
||||
src := readShippedPlugin(t, "quota.lua")
|
||||
if !strings.Contains(src, "string.match(u,") {
|
||||
t.Error("hostOf does not use string.match — string.gmatch yields whole " +
|
||||
"matches, so the capture group is ignored and the function returns nil")
|
||||
}
|
||||
|
||||
ps := quotaWithDecision(t)
|
||||
for _, c := range []struct{ in, want string }{
|
||||
{"https://qijiar.top/v1", "qijiar.top"},
|
||||
{"https://api.deepseek.com", "api.deepseek.com"},
|
||||
{"http://127.0.0.1:19900", "127.0.0.1:19900"},
|
||||
{"https://s2a.frankhan.top:14514/v1", "s2a.frankhan.top:14514"},
|
||||
} {
|
||||
out, err := ps.Call("quota", "__hostof", map[string]interface{}{"u": c.in})
|
||||
if err != nil {
|
||||
t.Fatalf("hostOf(%s): %v", c.in, err)
|
||||
}
|
||||
mm, _ := out.(map[string]interface{})
|
||||
got, _ := mm["h"].(string)
|
||||
if got != c.want {
|
||||
t.Errorf("hostOf(%q) = %q, want %q", c.in, got, c.want)
|
||||
}
|
||||
if got == "" {
|
||||
t.Errorf("hostOf(%q) returned nothing — every lookup would be "+
|
||||
"HOST_MEMO[nil]", c.in)
|
||||
}
|
||||
}
|
||||
|
||||
// The same host under two paths must produce the SAME key, which is the whole
|
||||
// reason the memo is keyed this way.
|
||||
out, err := ps.Call("quota", "__hostof", map[string]interface{}{"u": "https://x.example/a/v1"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
out2, err := ps.Call("quota", "__hostof", map[string]interface{}{"u": "https://x.example/b/v1"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m1, _ := out.(map[string]interface{})
|
||||
m2, _ := out2.(map[string]interface{})
|
||||
if m1["h"] != m2["h"] {
|
||||
t.Errorf("two paths on one host produced different keys: %v and %v — "+
|
||||
"they would never share a memo", m1["h"], m2["h"])
|
||||
}
|
||||
}
|
||||
|
||||
// --- helpers ---
|
||||
|
||||
// readShippedPlugin returns the bundled source, so the tests exercise the file
|
||||
// that actually ships rather than a copy that can drift from it.
|
||||
func readShippedPlugin(t *testing.T, name string) string {
|
||||
t.Helper()
|
||||
b, err := bundledAdapters.ReadFile("plugins/" + name)
|
||||
if err != nil {
|
||||
t.Fatalf("read embedded %s: %v", name, err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// quotaWithDecision loads quota.lua with its local backoffSeconds exposed as an
|
||||
// export. The plugin's own exports list is appended to, so the export allowlist is
|
||||
// satisfied exactly as it is in production.
|
||||
func quotaWithDecision(t *testing.T) *Plugins {
|
||||
t.Helper()
|
||||
src := readShippedPlugin(t, "quota.lua")
|
||||
shim := strings.Replace(src, "return PLUGIN",
|
||||
"PLUGIN.exports[#PLUGIN.exports+1] = '__decision'\n"+
|
||||
// Wrapped, not exported raw: backoffSeconds takes (at, streak)
|
||||
// positionally, and the test calls it with a table because that is the
|
||||
// shape a Go caller can pass. Exporting the raw function made every
|
||||
// assertion read `at = <the argument table>, streak = nil`, so the
|
||||
// function took its "no streak" path and returned false for every input
|
||||
// — which looked exactly like a backoff that never fires.
|
||||
"PLUGIN.__decision = function(q) return { r = backoffSeconds(q.at, q.streak) } end\n"+
|
||||
"PLUGIN.exports[#PLUGIN.exports+1] = '__hostof'\n"+
|
||||
"PLUGIN.__hostof = function(q) return { h = hostOf(q.u) } end\n"+
|
||||
"return PLUGIN", 1)
|
||||
if shim == src {
|
||||
t.Fatal("could not inject the decision shim — quota.lua no longer ends with 'return PLUGIN'")
|
||||
}
|
||||
return newQuotaPluginsWith(t, shim, "127.0.0.1")
|
||||
}
|
||||
|
||||
// quotaSkip asks the plugin whether a host that failed `streak` times and was
|
||||
// last heard from `age` seconds ago should be skipped.
|
||||
func quotaSkip(t *testing.T, ps *Plugins, streak, age int) bool {
|
||||
out, err := ps.Call("quota", "__decision", map[string]interface{}{
|
||||
"at": float64(time.Now().Unix() - int64(age)),
|
||||
"streak": float64(streak),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("decision: %v", err)
|
||||
}
|
||||
// A single-field table collapses to the bare value on encode, so both shapes
|
||||
// mean the same thing here.
|
||||
switch v := out.(type) {
|
||||
case bool:
|
||||
return v
|
||||
case map[string]interface{}:
|
||||
b, ok := v["r"].(bool)
|
||||
if !ok {
|
||||
t.Fatalf("decision returned %T, not a boolean: %#v", v["r"], v)
|
||||
}
|
||||
return b
|
||||
default:
|
||||
t.Fatalf("decision returned %T (%v), not a boolean", out, out)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// newQuotaPlugins boots the shipped quota.lua against a temp plugin dir.
|
||||
func newQuotaPlugins(t *testing.T, baseURL string) *Plugins {
|
||||
t.Helper()
|
||||
return newQuotaPluginsWith(t, readShippedPlugin(t, "quota.lua"), baseURL)
|
||||
}
|
||||
|
||||
func newQuotaPluginsWith(t *testing.T, src, baseURL string) *Plugins {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
pluginDir := filepath.Join(dir, "plugins")
|
||||
if err := os.MkdirAll(pluginDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
ps := NewPlugins(vm, pluginDir)
|
||||
t.Cleanup(ps.Close)
|
||||
ps.SetOutboundHosts([]string{"127.0.0.1"})
|
||||
if err := os.WriteFile(filepath.Join(pluginDir, "quota.lua"), []byte(src), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ps.LoadDir(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p := ps.find("quota")
|
||||
if p == nil {
|
||||
t.Fatal("quota plugin did not load")
|
||||
}
|
||||
if p.LoadError != "" {
|
||||
t.Fatalf("quota plugin failed to compile: %s", p.LoadError)
|
||||
}
|
||||
_ = baseURL
|
||||
return ps
|
||||
}
|
||||
|
||||
// countingServer answers 404 on every path and counts what it is asked, so a test
|
||||
// can tell "the plugin skipped" (no requests) from "the plugin probed and found
|
||||
// nothing" (requests, no data).
|
||||
//
|
||||
// It used to answer 200 with a non-balance body, and the commandcode adapter
|
||||
// accepted that — so the host was memoised as a SUCCESS and the negative path was
|
||||
// never reached. The test then failed for the right reason, and would have
|
||||
// passed for the wrong one had the assertion been looser.
|
||||
type countingServer struct {
|
||||
*httptest.Server
|
||||
hits int
|
||||
}
|
||||
|
||||
func (c *countingServer) Hits() int { return c.hits }
|
||||
|
||||
func newCountingServer(t *testing.T) *countingServer {
|
||||
t.Helper()
|
||||
cs := &countingServer{}
|
||||
cs.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
cs.hits++
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
_, _ = w.Write([]byte(`{"error":"not found"}`))
|
||||
}))
|
||||
t.Cleanup(cs.Close)
|
||||
return cs
|
||||
}
|
||||
|
||||
// TestQuotaDueSourcesAndMemoRecordCoverTheShardGap pins the split that makes the
|
||||
// backoff work at all under sharding.
|
||||
//
|
||||
// The conflict, measured on a live 20-source gateway:
|
||||
// - probing is network-bound and Plugins.Call holds the plugin mutex for the
|
||||
// whole Lua call, so /call/ serialises (29s)
|
||||
// - so the page uses ShardCall — and a shard has its own PLUGIN.state
|
||||
// - and the backoff lives in PLUGIN.state: fanout left the memo at 0 entries,
|
||||
// so dead hosts were re-probed forever
|
||||
//
|
||||
// The resolution is to put the DECISION and the RECORDING back on the shared VM
|
||||
// (microseconds, no network) and shard only the network wait. These two exports
|
||||
// are that seam, and without them the fanout path silently defeats the backoff.
|
||||
func TestQuotaDueSourcesAndMemoRecordCoverTheShardGap(t *testing.T) {
|
||||
src := readShippedPlugin(t, "quota.lua")
|
||||
for _, want := range []string{"due_sources", "memo_record"} {
|
||||
if !strings.Contains(src, want) {
|
||||
t.Errorf("quota.lua does not export %s — a sharded pass cannot see the "+
|
||||
"backoff state, nor leave any behind", want)
|
||||
}
|
||||
if !strings.Contains(src, `"`+want+`"`) {
|
||||
t.Errorf("%s is not listed in PLUGIN.exports, so the page cannot call it", want)
|
||||
}
|
||||
}
|
||||
// due_sources must not probe. If it did, the whole point of asking it first
|
||||
// would be lost: the cost would be paid before any shard runs.
|
||||
if strings.Contains(src, "due_sources = function(payload)") &&
|
||||
strings.Contains(src, "due_sources") {
|
||||
// The body is small; check it does not call collectSource.
|
||||
i := strings.Index(src, "due_sources = function(payload)")
|
||||
j := strings.Index(src[i:], "\nend\n")
|
||||
if j > 0 {
|
||||
body := src[i : i+j]
|
||||
if strings.Contains(body, "collectSource(") {
|
||||
t.Error("due_sources calls collectSource — the filter must make no " +
|
||||
"network request, or it costs the same as probing everything")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaMemoRecordGrowsTheStreakFromShards covers the recording half: a host
|
||||
// that failed inside a shard must end up in the SHARED memo with a streak, or the
|
||||
// next refresh has nothing to back off from.
|
||||
func TestQuotaMemoRecordGrowsTheStreakFromShards(t *testing.T) {
|
||||
srv := newCountingServer(t)
|
||||
ps := newQuotaPlugins(t, srv.URL)
|
||||
host := hostOnly(srv.URL)
|
||||
|
||||
read := func() map[string]interface{} {
|
||||
st, _ := ps.State("quota").(map[string]interface{})
|
||||
memo, _ := st["probe_memo"].(map[string]interface{})
|
||||
return memo
|
||||
}
|
||||
|
||||
// A shard reports a host that answered with nothing.
|
||||
for i := 1; i <= 3; i++ {
|
||||
out, err := ps.Call("quota", "memo_record", map[string]interface{}{
|
||||
"results": []interface{}{
|
||||
map[string]interface{}{"host": srv.URL + "/v1"},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("memo_record %d: %v", i, err)
|
||||
}
|
||||
if out != true {
|
||||
t.Fatalf("memo_record returned %#v", out)
|
||||
}
|
||||
}
|
||||
memo := read()
|
||||
e, ok := memo[host].(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("memo has no entry for %s: %#v", host, memo)
|
||||
}
|
||||
if got, _ := e["streak"].(float64); got != 3 {
|
||||
t.Errorf("streak after three shard failures = %.0f, want 3", got)
|
||||
}
|
||||
|
||||
// A host that answers is recorded as a positive memo, which resets the streak.
|
||||
if _, err := ps.Call("quota", "memo_record", map[string]interface{}{
|
||||
"results": []interface{}{
|
||||
map[string]interface{}{"host": srv.URL + "/v1", "vendor": "deepseek"},
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := read()[host]; got != "deepseek" {
|
||||
t.Errorf("a host that answered is memoised as %#v, want the vendor name — "+
|
||||
"the streak must not survive a success", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaDueSourcesSkipsBackedOffHosts is the behaviour the page depends on:
|
||||
// the filter must keep a backed-off host out of the sharded set.
|
||||
func TestQuotaDueSourcesSkipsBackedOffHosts(t *testing.T) {
|
||||
srv := newCountingServer(t)
|
||||
ps := newQuotaPlugins(t, srv.URL)
|
||||
host := hostOnly(srv.URL)
|
||||
|
||||
// One failure, which is enough to put the host inside its window.
|
||||
if _, err := ps.Call("quota", "memo_record", map[string]interface{}{
|
||||
"results": []interface{}{map[string]interface{}{"host": srv.URL + "/v1"}},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// A SECOND host, so the two cases are genuinely independent. The memo is keyed
|
||||
// by host, so two paths on one host are one entry — using the same server here
|
||||
// would have asserted that a per-host memo behaves per-path.
|
||||
other := newCountingServer(t)
|
||||
srcs := []interface{}{
|
||||
map[string]interface{}{"name": "dead", "base_url": srv.URL + "/v1", "api_key": "k"},
|
||||
map[string]interface{}{"name": "fresh", "base_url": other.URL + "/v1", "api_key": "k"},
|
||||
}
|
||||
out, err := ps.Call("quota", "due_sources", map[string]interface{}{"sources": srcs})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, ok := out.(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("due_sources returned %#v", out)
|
||||
}
|
||||
due, _ := m["due"].([]interface{})
|
||||
backing, _ := m["backing_off"].([]interface{})
|
||||
if len(backing) != 1 {
|
||||
t.Errorf("backing_off = %#v, want exactly the host that failed", backing)
|
||||
} else if b, _ := backing[0].(map[string]interface{}); b["host"] != host {
|
||||
t.Errorf("backing off %v, want %s — the key must be the host, or it will "+
|
||||
"never match on the next refresh", b["host"], host)
|
||||
}
|
||||
if len(due) != 1 {
|
||||
t.Errorf("due = %#v, want only the unprobed host", due)
|
||||
} else if d, _ := due[0].(map[string]interface{}); d["name"] != "fresh" {
|
||||
t.Errorf("due contains %v, want fresh", d["name"])
|
||||
}
|
||||
|
||||
// And it must make no network request — the whole saving depends on it.
|
||||
if n := srv.Hits(); n != 0 {
|
||||
t.Errorf("due_sources made %d requests; it must decide without probing", n)
|
||||
}
|
||||
}
|
||||
|
||||
// hostOnly mirrors the plugin's own hostOf. Used only to CHECK what the plugin
|
||||
// wrote — seeding a memo under a different key produces a memo the plugin never
|
||||
// looks up, which cost this file a couple of rounds before it was noticed.
|
||||
func hostOnly(baseURL string) string {
|
||||
u := strings.TrimPrefix(strings.TrimPrefix(baseURL, "https://"), "http://")
|
||||
if i := strings.Index(u, "/"); i >= 0 {
|
||||
u = u[:i]
|
||||
}
|
||||
return u
|
||||
}
|
||||
Reference in New Issue
Block a user