1 // Package actor provides generic building blocks for CSP/actor-model
2 // concurrency in Go. Each type maps to a common inter-goroutine
3 // communication pattern, eliminating the per-actor boilerplate of
4 // defining request structs, response channels, and caller methods.
5 //
6 // The five channel types cover every pattern observed in production
7 // actor codebases:
8 //
9 // - [Func] synchronous request -> response (function call)
10 // - [Query] synchronous void -> response (getter)
11 // - [Proc] synchronous request -> done (setter / command)
12 // - [Signal] synchronous void -> done (trigger / reset)
13 // - [Inbox] async fire-and-forget (logging / notifications)
14 //
15 // [Lifecycle] manages the stop/done shutdown protocol.
16 //
17 // All synchronous types use unbuffered channels. The caller blocks
18 // until the actor processes the message. This eliminates the class of
19 // bugs where a buffered write appears to succeed but the actor hasn't
20 // seen it yet (the transport manager race pattern).
21 //
22 // # Usage
23 //
24 // Define an actor struct embedding the channel types it needs:
25 //
26 // type Counter struct {
27 // inc actor.Signal
28 // get actor.Query[int]
29 // actor.Lifecycle
30 // }
31 //
32 // Initialize in the constructor:
33 //
34 // func NewCounter() *Counter {
35 // c := &Counter{
36 // inc: actor.NewSignal(),
37 // get: actor.NewQuery[int](),
38 // Lifecycle: actor.NewLifecycle(),
39 // }
40 // actor.Go(c.Lifecycle, c.run)
41 // return c
42 // }
43 //
44 // Write the actor loop using Recv() in select:
45 //
46 // func (c *Counter) run() {
47 // var n int
48 // for {
49 // select {
50 // case <-c.Stopping():
51 // return
52 // case msg := <-c.inc.Recv():
53 // n++
54 // msg.Done()
55 // case msg := <-c.get.Recv():
56 // msg.Reply(n)
57 // }
58 // }
59 // }
60 //
61 // Public methods are one-liners:
62 //
63 // func (c *Counter) Inc() { c.inc.Call() }
64 // func (c *Counter) Get() int { return c.get.Call() }
65 //
66 // # Design Principles
67 //
68 // Every interaction between peers requires independent bidirectional
69 // channels. Simplex+lock is strictly more complex and introduces
70 // deadlock risk.
71 //
72 // Backpressure is expressed by channel state, not by blocking the
73 // sender arbitrarily. Neither side should be able to freeze the other
74 // - use [Inbox] with a bounded buffer when the sender must not block.
75 //
76 // State ownership stays with the actor goroutine. Short-lived callers
77 // get copies (via response channels), not references to internal state.
78 // Death of a worker is reported via [Lifecycle], not hidden or
79 // auto-recovered.
80 package actor
81 82 import "context"
83 84 // ---------------------------------------------------------------------------
85 // Func: synchronous request -> response
86 // ---------------------------------------------------------------------------
87 88 // Func is a synchronous function call channel. The caller sends a
89 // request of type Req and blocks until the actor replies with Resp.
90 type Func[Req, Resp any] struct {
91 ch chan FuncMsg[Req, Resp]
92 }
93 94 // FuncMsg is the message received by the actor from a [Func] channel.
95 // Access the request via the Req field; send the response via Reply.
96 type FuncMsg[Req, Resp any] struct {
97 Req Req
98 resp chan Resp
99 }
100 101 // NewFunc creates a Func channel (unbuffered).
102 func NewFunc[Req, Resp any]() Func[Req, Resp] {
103 return Func[Req, Resp]{ch: make(chan FuncMsg[Req, Resp])}
104 }
105 106 // Call sends req to the actor and blocks for the response.
107 func (f Func[Req, Resp]) Call(req Req) Resp {
108 resp := make(chan Resp, 1)
109 f.ch <- FuncMsg[Req, Resp]{Req: req, resp: resp}
110 return <-resp
111 }
112 113 // CallCtx sends req to the actor and blocks for the response, but
114 // abandons the call if ctx is cancelled. Returns ctx.Err() on cancel.
115 // If the actor has already received the request when the caller bails,
116 // the actor's Reply writes to a buffered channel that gets GC'd.
117 func (f Func[Req, Resp]) CallCtx(ctx context.Context, req Req) (Resp, error) {
118 resp := make(chan Resp, 1)
119 select {
120 case f.ch <- FuncMsg[Req, Resp]{Req: req, resp: resp}:
121 case <-ctx.Done():
122 var zero Resp
123 return zero, ctx.Err()
124 }
125 select {
126 case r := <-resp:
127 return r, nil
128 case <-ctx.Done():
129 var zero Resp
130 return zero, ctx.Err()
131 }
132 }
133 134 // Recv returns the receive-only channel for use in select.
135 func (f Func[Req, Resp]) Recv() <-chan FuncMsg[Req, Resp] {
136 return f.ch
137 }
138 139 // Reply sends the response back to the caller.
140 // Must be called exactly once per received message.
141 func (m FuncMsg[Req, Resp]) Reply(r Resp) {
142 m.resp <- r
143 }
144 145 // ---------------------------------------------------------------------------
146 // Query: synchronous void -> response
147 // ---------------------------------------------------------------------------
148 149 // Query is a synchronous getter channel. The caller sends no data
150 // and blocks until the actor replies with Resp.
151 type Query[Resp any] struct {
152 ch chan QueryMsg[Resp]
153 }
154 155 // QueryMsg is the message received by the actor from a [Query] channel.
156 type QueryMsg[Resp any] struct {
157 resp chan Resp
158 }
159 160 // NewQuery creates a Query channel (unbuffered).
161 func NewQuery[Resp any]() Query[Resp] {
162 return Query[Resp]{ch: make(chan QueryMsg[Resp])}
163 }
164 165 // Call sends a query and blocks for the response.
166 func (q Query[Resp]) Call() Resp {
167 resp := make(chan Resp, 1)
168 q.ch <- QueryMsg[Resp]{resp: resp}
169 return <-resp
170 }
171 172 // CallCtx sends a query and blocks for the response, but abandons
173 // the call if ctx is cancelled.
174 func (q Query[Resp]) CallCtx(ctx context.Context) (Resp, error) {
175 resp := make(chan Resp, 1)
176 select {
177 case q.ch <- QueryMsg[Resp]{resp: resp}:
178 case <-ctx.Done():
179 var zero Resp
180 return zero, ctx.Err()
181 }
182 select {
183 case r := <-resp:
184 return r, nil
185 case <-ctx.Done():
186 var zero Resp
187 return zero, ctx.Err()
188 }
189 }
190 191 // Recv returns the receive-only channel for use in select.
192 func (q Query[Resp]) Recv() <-chan QueryMsg[Resp] {
193 return q.ch
194 }
195 196 // Reply sends the response back to the caller.
197 func (m QueryMsg[Resp]) Reply(r Resp) {
198 m.resp <- r
199 }
200 201 // ---------------------------------------------------------------------------
202 // Proc: synchronous request -> done
203 // ---------------------------------------------------------------------------
204 205 // Proc is a synchronous command channel. The caller sends a request
206 // and blocks until the actor confirms it has been processed.
207 type Proc[Req any] struct {
208 ch chan ProcMsg[Req]
209 }
210 211 // ProcMsg is the message received by the actor from a [Proc] channel.
212 // Access the request via the Req field; signal completion via Done.
213 type ProcMsg[Req any] struct {
214 Req Req
215 done chan struct{}
216 }
217 218 // NewProc creates a Proc channel (unbuffered).
219 func NewProc[Req any]() Proc[Req] {
220 return Proc[Req]{ch: make(chan ProcMsg[Req])}
221 }
222 223 // Call sends req to the actor and blocks until it is processed.
224 func (p Proc[Req]) Call(req Req) {
225 done := make(chan struct{})
226 p.ch <- ProcMsg[Req]{Req: req, done: done}
227 <-done
228 }
229 230 // CallCtx sends req to the actor and blocks until processed, but
231 // abandons the call if ctx is cancelled.
232 func (p Proc[Req]) CallCtx(ctx context.Context, req Req) error {
233 done := make(chan struct{})
234 select {
235 case p.ch <- ProcMsg[Req]{Req: req, done: done}:
236 case <-ctx.Done():
237 return ctx.Err()
238 }
239 select {
240 case <-done:
241 return nil
242 case <-ctx.Done():
243 return ctx.Err()
244 }
245 }
246 247 // Recv returns the receive-only channel for use in select.
248 func (p Proc[Req]) Recv() <-chan ProcMsg[Req] {
249 return p.ch
250 }
251 252 // Done signals the caller that the request has been processed.
253 // Must be called exactly once per received message.
254 func (m ProcMsg[Req]) Done() {
255 close(m.done)
256 }
257 258 // ---------------------------------------------------------------------------
259 // Signal: synchronous void -> done
260 // ---------------------------------------------------------------------------
261 262 // Signal is a synchronous trigger channel. The caller sends no data
263 // and blocks until the actor confirms it has handled the signal.
264 type Signal struct {
265 ch chan SignalMsg
266 }
267 268 // SignalMsg is the message received by the actor from a [Signal] channel.
269 type SignalMsg struct {
270 done chan struct{}
271 }
272 273 // NewSignal creates a Signal channel (unbuffered).
274 func NewSignal() Signal {
275 return Signal{ch: make(chan SignalMsg)}
276 }
277 278 // Call sends the signal and blocks until it is processed.
279 func (s Signal) Call() {
280 done := make(chan struct{})
281 s.ch <- SignalMsg{done: done}
282 <-done
283 }
284 285 // CallCtx sends the signal and blocks until processed, but abandons
286 // the call if ctx is cancelled.
287 func (s Signal) CallCtx(ctx context.Context) error {
288 done := make(chan struct{})
289 select {
290 case s.ch <- SignalMsg{done: done}:
291 case <-ctx.Done():
292 return ctx.Err()
293 }
294 select {
295 case <-done:
296 return nil
297 case <-ctx.Done():
298 return ctx.Err()
299 }
300 }
301 302 // Recv returns the receive-only channel for use in select.
303 func (s Signal) Recv() <-chan SignalMsg {
304 return s.ch
305 }
306 307 // Done signals the caller that the signal has been handled.
308 func (m SignalMsg) Done() {
309 close(m.done)
310 }
311 312 // ---------------------------------------------------------------------------
313 // Inbox: async fire-and-forget
314 // ---------------------------------------------------------------------------
315 316 // Inbox is a buffered channel for fire-and-forget messages.
317 // Senders do not wait for the actor to process the message.
318 // Use for logging, metrics, and other non-critical notifications.
319 type Inbox[T any] struct {
320 ch chan T
321 }
322 323 // NewInbox creates an Inbox with the given buffer size.
324 // A buffer of 0 creates an unbuffered (synchronous) inbox.
325 func NewInbox[T any](buffer int) Inbox[T] {
326 if buffer < 0 {
327 buffer = 0
328 }
329 return Inbox[T]{ch: make(chan T, buffer)}
330 }
331 332 // Send enqueues a message. Blocks if the buffer is full.
333 func (i Inbox[T]) Send(v T) {
334 i.ch <- v
335 }
336 337 // TrySend attempts to enqueue a message without blocking.
338 // Returns false if the buffer is full (message dropped).
339 func (i Inbox[T]) TrySend(v T) bool {
340 select {
341 case i.ch <- v:
342 return true
343 default:
344 return false
345 }
346 }
347 348 // Recv returns the receive-only channel for use in select.
349 func (i Inbox[T]) Recv() <-chan T {
350 return i.ch
351 }
352 353 // ---------------------------------------------------------------------------
354 // Lifecycle: stop/done shutdown protocol
355 // ---------------------------------------------------------------------------
356 357 // Lifecycle manages the shutdown handshake between an actor and its
358 // owner. The owner calls Stop to request shutdown and wait for
359 // completion. The actor calls MarkDone (typically via defer) when it
360 // has finished.
361 type Lifecycle struct {
362 stop chan struct{}
363 done chan struct{}
364 }
365 366 // NewLifecycle creates an initialized Lifecycle.
367 func NewLifecycle() Lifecycle {
368 return Lifecycle{
369 stop: make(chan struct{}),
370 done: make(chan struct{}),
371 }
372 }
373 374 // Stop signals the actor to stop and blocks until it finishes.
375 // Must be called at most once.
376 func (l Lifecycle) Stop() {
377 close(l.stop)
378 <-l.done
379 }
380 381 // Stopping returns a channel that closes when Stop is called.
382 // Use in select to detect shutdown requests.
383 func (l Lifecycle) Stopping() <-chan struct{} {
384 return l.stop
385 }
386 387 // Stopped returns a channel that closes when the actor finishes.
388 // Useful for external observers waiting on actor completion.
389 func (l Lifecycle) Stopped() <-chan struct{} {
390 return l.done
391 }
392 393 // MarkDone signals that the actor goroutine has finished.
394 // Call via defer at the top of the actor function.
395 func (l Lifecycle) MarkDone() {
396 close(l.done)
397 }
398 399 // Go starts fn in a new goroutine and automatically calls
400 // l.MarkDone when fn returns.
401 func Go(l Lifecycle, fn func()) {
402 go func() {
403 defer l.MarkDone()
404 fn()
405 }()
406 }
407