Skip to content

Commit 23d786a

Browse files
committed
feat(distributed): add per-request node ID context holder
Introduce pkg/distributedhdr, a leaf package carrying a per-request *atomic.Value holder for the picked worker node ID from the SmartRouter (core/services/nodes) up to the HTTP response writer wrapper (core/http/middleware). Avoids the import cycle that a shared key in either consumer would create. Exposes NewHolder, WithHolder, Holder, Stamp, Load, Inherit. The holder is atomic.Value so cross-goroutine publish from the router to the response writer wrapper is race-clean. Assisted-by: Claude:claude-opus-4-7[1m] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
1 parent 90ea327 commit 23d786a

2 files changed

Lines changed: 207 additions & 0 deletions

File tree

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
// Package distributedhdr carries a per-request "which worker node served
2+
// me" record from the distributed router (core/services/nodes) up to the
3+
// HTTP response writer wrapper (core/http/middleware). It is the bridge
4+
// for the X-LocalAI-Node response header without coupling those two
5+
// packages directly or going through any shared mutable state.
6+
//
7+
// Why its own package: both core/http/middleware and core/services/nodes
8+
// already import pkg/model, and the natural homes for this key would
9+
// create an import cycle if either side hosted the helper. Putting the
10+
// key and the tiny helpers here keeps it neutral - producer and consumer
11+
// import a leaf package, not each other.
12+
package distributedhdr
13+
14+
import (
15+
"context"
16+
"sync/atomic"
17+
)
18+
19+
// ctxKey is an unexported context-key type so external packages cannot
20+
// collide with our key by accident.
21+
type ctxKey struct{}
22+
23+
// holderKey is the singleton context key used by WithHolder / Holder /
24+
// Stamp. Unexported on purpose: producers and consumers must go through
25+
// the helpers below so the storage representation stays an
26+
// implementation detail.
27+
var holderKey = ctxKey{}
28+
29+
// NewHolder returns an empty holder ready to be attached to a request
30+
// context via WithHolder. The middleware creates one per HTTP request;
31+
// the router fills it; the response writer wrapper reads it on the
32+
// first byte. The atomic.Value gives us race-clean publish across the
33+
// goroutines that may write (router) and read (response writer
34+
// wrapper) the slot.
35+
func NewHolder() *atomic.Value {
36+
return &atomic.Value{}
37+
}
38+
39+
// WithHolder returns a derived context that carries the provided holder.
40+
// The middleware calls this on the per-request context BEFORE the
41+
// downstream handler chain runs, so any goroutine that inherits this
42+
// context (notably the SmartRouter / ModelRouterAdapter) can find the
43+
// holder via Stamp.
44+
func WithHolder(ctx context.Context, h *atomic.Value) context.Context {
45+
if h == nil {
46+
return ctx
47+
}
48+
return context.WithValue(ctx, holderKey, h)
49+
}
50+
51+
// Holder returns the holder attached to ctx, or nil if none was set.
52+
// Callers that just want to publish should prefer Stamp; Holder is
53+
// useful for tests and for propagating the holder across derived
54+
// contexts (see Inherit).
55+
func Holder(ctx context.Context) *atomic.Value {
56+
if ctx == nil {
57+
return nil
58+
}
59+
h, _ := ctx.Value(holderKey).(*atomic.Value)
60+
return h
61+
}
62+
63+
// Stamp records the picked worker node ID into the holder attached to
64+
// ctx. No-op when:
65+
// - ctx is nil
66+
// - no holder is attached (e.g. the X-LocalAI-Node feature is off, so
67+
// the middleware never ran)
68+
// - nodeID is empty
69+
//
70+
// Stamp is safe to call from any goroutine. Subsequent reads via Load
71+
// observe the most recent stamp.
72+
func Stamp(ctx context.Context, nodeID string) {
73+
if nodeID == "" {
74+
return
75+
}
76+
h := Holder(ctx)
77+
if h == nil {
78+
return
79+
}
80+
h.Store(nodeID)
81+
}
82+
83+
// Load returns the node ID most recently stamped into the holder, or
84+
// "" if nothing has been stamped. Intended for the response writer
85+
// wrapper's first-byte hook.
86+
func Load(h *atomic.Value) string {
87+
if h == nil {
88+
return ""
89+
}
90+
v, _ := h.Load().(string)
91+
return v
92+
}
93+
94+
// Inherit copies the holder reference from src into dst when present.
95+
// Used at request-handling seams where the request context is replaced
96+
// with a fresh context derived from the long-lived application context
97+
// (so the cancel semantics of the original context are preserved while
98+
// also allowing the load path to outlive the HTTP transport). The
99+
// holder is a pointer, so both contexts point at the same slot; a
100+
// router stamp via Stamp(dst, ...) is observed by the middleware that
101+
// reads through src's holder.
102+
func Inherit(dst, src context.Context) context.Context {
103+
h := Holder(src)
104+
if h == nil {
105+
return dst
106+
}
107+
return WithHolder(dst, h)
108+
}
Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
package distributedhdr_test
2+
3+
import (
4+
"context"
5+
"sync"
6+
"testing"
7+
8+
"github.com/mudler/LocalAI/pkg/distributedhdr"
9+
. "github.com/onsi/ginkgo/v2"
10+
. "github.com/onsi/gomega"
11+
)
12+
13+
func TestDistributedHdr(t *testing.T) {
14+
RegisterFailHandler(Fail)
15+
RunSpecs(t, "distributedhdr suite")
16+
}
17+
18+
var _ = Describe("distributedhdr", func() {
19+
It("returns empty string when no holder is attached", func() {
20+
Expect(distributedhdr.Holder(context.Background())).To(BeNil())
21+
Expect(distributedhdr.Load(distributedhdr.Holder(context.Background()))).To(BeEmpty())
22+
})
23+
24+
It("is a no-op when stamping into a context without a holder", func() {
25+
Expect(func() {
26+
distributedhdr.Stamp(context.Background(), "worker-x")
27+
}).ToNot(Panic())
28+
})
29+
30+
It("is a no-op when stamping a nil context", func() {
31+
Expect(func() {
32+
distributedhdr.Stamp(nil, "worker-x")
33+
}).ToNot(Panic())
34+
})
35+
36+
It("stores and retrieves a node ID through Stamp/Load", func() {
37+
h := distributedhdr.NewHolder()
38+
ctx := distributedhdr.WithHolder(context.Background(), h)
39+
40+
distributedhdr.Stamp(ctx, "worker-a")
41+
Expect(distributedhdr.Load(h)).To(Equal("worker-a"))
42+
})
43+
44+
It("retains the latest stamp when called twice", func() {
45+
h := distributedhdr.NewHolder()
46+
ctx := distributedhdr.WithHolder(context.Background(), h)
47+
48+
distributedhdr.Stamp(ctx, "worker-a")
49+
distributedhdr.Stamp(ctx, "worker-b")
50+
Expect(distributedhdr.Load(h)).To(Equal("worker-b"))
51+
})
52+
53+
It("does not overwrite when stamping with an empty string", func() {
54+
h := distributedhdr.NewHolder()
55+
ctx := distributedhdr.WithHolder(context.Background(), h)
56+
57+
distributedhdr.Stamp(ctx, "worker-a")
58+
distributedhdr.Stamp(ctx, "")
59+
Expect(distributedhdr.Load(h)).To(Equal("worker-a"))
60+
})
61+
62+
It("propagates the holder via Inherit into a derived context", func() {
63+
h := distributedhdr.NewHolder()
64+
src := distributedhdr.WithHolder(context.Background(), h)
65+
dst := distributedhdr.Inherit(context.Background(), src)
66+
67+
distributedhdr.Stamp(dst, "worker-c")
68+
Expect(distributedhdr.Load(h)).To(Equal("worker-c"))
69+
})
70+
71+
It("Inherit is a no-op when src has no holder", func() {
72+
dst := distributedhdr.Inherit(context.Background(), context.Background())
73+
Expect(distributedhdr.Holder(dst)).To(BeNil())
74+
})
75+
76+
It("publishes stamps across goroutines race-free", func() {
77+
h := distributedhdr.NewHolder()
78+
ctx := distributedhdr.WithHolder(context.Background(), h)
79+
80+
const N = 64
81+
var wg sync.WaitGroup
82+
wg.Add(N)
83+
for i := 0; i < N; i++ {
84+
i := i
85+
go func() {
86+
defer wg.Done()
87+
if i%2 == 0 {
88+
distributedhdr.Stamp(ctx, "worker-even")
89+
} else {
90+
distributedhdr.Stamp(ctx, "worker-odd")
91+
}
92+
}()
93+
}
94+
wg.Wait()
95+
96+
got := distributedhdr.Load(h)
97+
Expect(got).To(SatisfyAny(Equal("worker-even"), Equal("worker-odd")))
98+
})
99+
})

0 commit comments

Comments
 (0)