From 90415aaa043875099ea395d5612aae8292b09f5d Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:48:42 -0700 Subject: [PATCH 1/6] actor: add new actor package as distinct sub-module --- actor/go.mod | 19 +++++++++++++++++++ actor/go.sum | 29 +++++++++++++++++++++++++++++ actor/log.go | 12 ++++++++++++ 3 files changed, 60 insertions(+) create mode 100644 actor/go.mod create mode 100644 actor/go.sum create mode 100644 actor/log.go diff --git a/actor/go.mod b/actor/go.mod new file mode 100644 index 000000000..6bc5cd9e0 --- /dev/null +++ b/actor/go.mod @@ -0,0 +1,19 @@ +module github.com/lightningnetwork/lnd/actor + +go 1.24.9 + +require ( + github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084 + github.com/lightningnetwork/lnd/fn/v2 v2.0.8 + github.com/stretchr/testify v1.8.1 + pgregory.net/rapid v1.2.0 +) + +require ( + github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + golang.org/x/exp v0.0.0-20231226003508-02704c960a9b // indirect + golang.org/x/sync v0.7.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/actor/go.sum b/actor/go.sum new file mode 100644 index 000000000..bcd10dc1f --- /dev/null +++ b/actor/go.sum @@ -0,0 +1,29 @@ +github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c h1:4HxD1lBUGUddhzgaNgrCPsFWd7cGYNpeFUgd9ZIgyM0= +github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c/go.mod h1:w7xnGOhwT3lmrS4H3b/D1XAXxvh+tbhUm8xeHN2y3TQ= +github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084 h1:y3bvkt8ki0KX35eUEU8XShRHusz1S+55QwXUTmxn888= +github.com/btcsuite/btclog/v2 v2.0.1-0.20250602222548-9967d19bb084/go.mod h1:XItGUfVOxotJL8kkuk2Hj3EVow5KCugXl3wWfQ6K0AE= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/lightningnetwork/lnd/fn/v2 v2.0.8 h1:r2SLz7gZYQPVc3IZhU82M66guz3Zk2oY+Rlj9QN5S3g= +github.com/lightningnetwork/lnd/fn/v2 v2.0.8/go.mod h1:TOzwrhjB/Azw1V7aa8t21ufcQmdsQOQMDtxVOQWNl8s= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +golang.org/x/exp v0.0.0-20231226003508-02704c960a9b h1:kLiC65FbiHWFAOu+lxwNPujcsl8VYyTYYEZnsOO1WK4= +golang.org/x/exp v0.0.0-20231226003508-02704c960a9b/go.mod h1:iRJReGqOEeBhDZGkGbynYwcHlctCvnjTYIamk7uXpHI= +golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= +golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= +pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= diff --git a/actor/log.go b/actor/log.go new file mode 100644 index 000000000..77da9070d --- /dev/null +++ b/actor/log.go @@ -0,0 +1,12 @@ +package actor + +import "github.com/btcsuite/btclog/v2" + +// log is a logger that is initialized as disabled. This means the package will +// not perform any logging by default until a logger is set. +var log = btclog.Disabled + +// UseLogger uses a specified Logger to output package logging info. +func UseLogger(logger btclog.Logger) { + log = logger +} From 3069829d613de68f0bdf56b21c8186dc02debdf1 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:51:54 -0700 Subject: [PATCH 2/6] actor: add Future[T] and Promise[T] w/ concrete impls In this commit, we add two new fundamental data structures: Future[T] and Promise[T]. A future is a response that might be ready at some point in the future. This is already a common pattern in Go, we just make a type safe wrapper around the typical operations: block w/ a timeout, add a call back for execution, pipeline the response to a new future. A promise is an intent to complete a future. Typically the caller receives the future, and the callee is able to complete the future using a promise. --- actor/future.go | 158 +++++++++++++++ actor/future_test.go | 467 +++++++++++++++++++++++++++++++++++++++++++ actor/interface.go | 46 +++++ 3 files changed, 671 insertions(+) create mode 100644 actor/future.go create mode 100644 actor/future_test.go create mode 100644 actor/interface.go diff --git a/actor/future.go b/actor/future.go new file mode 100644 index 000000000..8c211691f --- /dev/null +++ b/actor/future.go @@ -0,0 +1,158 @@ +package actor + +import ( + "context" + "sync" + "sync/atomic" + + "github.com/lightningnetwork/lnd/fn/v2" +) + +// promiseImpl is a structure that can be used to complete a Future. It provides +// methods to set the result of an asynchronous operation and to obtain the +// Future interface for consumers. +// The promiseImpl itself is not typically exposed directly to consumers of the +// future's result; they interact with the Future interface. +type promiseImpl[T any] struct { + fut *futureImpl[T] +} + +// NewPromise creates a new Promise. The associated Future, which consumers can +// use to await the result, can be obtained via the Future() method. The Future +// is completed by calling the Complete() method on this Promise. +func NewPromise[T any]() Promise[T] { + return &promiseImpl[T]{ + fut: &futureImpl[T]{ + // done is a channel that will be closed when the future + // is completed. + done: make(chan struct{}), + }, + } +} + +// Future returns the Future interface associated with this Promise. Consumers +// can use this to Await the result or register callbacks. +func (p *promiseImpl[T]) Future() Future[T] { + return p.fut +} + +// Complete attempts to set the result of the future. It returns true if this +// call successfully set the result (i.e., it was the first to complete it), +// and false if the future had already been completed. This ensures that a +// future can only be completed once. The completion involves storing the result +// and signaling any goroutines waiting on the future's done channel. +func (p *promiseImpl[T]) Complete(result fn.Result[T]) bool { + var success bool + p.fut.completeOnce.Do(func() { + p.fut.resultCache.Store(&result) + close(p.fut.done) + + success = true + }) + + return success +} + +// futureImpl is the concrete implementation of the Future interface. It manages +// the state of an asynchronous computation's result. +type futureImpl[T any] struct { + // resultCache stores the fn.Result[T] after the future is completed. + // It's of type atomic.Pointer to allow lock-free reads after completion + // with improved type safety over atomic.Value. + resultCache atomic.Pointer[fn.Result[T]] + + // done is closed once the future is completed, signaling any waiting + // Await calls. + done chan struct{} + + // completeOnce ensures that the logic to set the result and close the + // done channel is executed only once. + completeOnce sync.Once +} + +// Await blocks until the result is available or the passed context is +// cancelled. If the future is already completed, it returns the result +// immediately. Otherwise, it waits for either the future's completion or the +// context's cancellation. +func (f *futureImpl[T]) Await(ctx context.Context) fn.Result[T] { + // First, try a non-blocking load from the cache. If the future is + // already completed, this will return the result directly. + if resPtr := f.resultCache.Load(); resPtr != nil { + return *resPtr + } + + // Wait for either the future to be done or the context to be cancelled. + select { + case <-f.done: + // The future has been completed. Load the result from the + // cache. It must be present now. Load and dereference. + // This load is safe because the 'done' channel is closed only + // after the resultCache is written (ensured by completeOnce). + resPtr := f.resultCache.Load() + + // resPtr should not be nil here as <-f.done was signaled. + return *resPtr + + case <-ctx.Done(): + // The waiting context was cancelled before the future completed. + return fn.Err[T](ctx.Err()) + } +} + +// ThenApply registers a function to transform the result of a future. The +// original future is not modified; a new Future instance representing the +// transformed result is returned. Once the original future completes +// successfully, the provided transformation function (fApply) is called with +// the result. The transformation is applied asynchronously in a new goroutine. +// If the passed context is cancelled while waiting for the +// original future to complete, the returned future will yield the context's +// error. +func (f *futureImpl[T]) ThenApply(ctx context.Context, + fApply func(T) T) Future[T] { + + // Create a new promise for the transformed result. + transformedPromise := NewPromise[T]() + + go func() { + // Await the original future's result, respecting the passed + // context for cancellation. + originalResult := f.Await(ctx) + + // If the original future completed with an error (or Await was + // cancelled by its context), complete the transformed future + // with the same error. + // This also handles the case where originalResult.Await(ctx) + // itself returned ctx.Err(). + if originalResult.IsErr() { + transformedPromise.Complete(originalResult) + return + } + + // Otherwise, the original future completed successfully. Apply the + // transformation function to its result. + originalResult.WhenOk(func(res T) { + newValue := fApply(res) + transformedPromise.Complete(fn.Ok(newValue)) + }) + }() + + return transformedPromise.Future() +} + +// OnComplete registers a function to be called when the result is ready. If the +// passed context is cancelled before the future completes, the callback +// function (cFunc) will be invoked with the context's error. The callback is +// executed in a new goroutine, so it does not block the completion path of the +// original future. +func (f *futureImpl[T]) OnComplete(ctx context.Context, + cFunc func(fn.Result[T])) { + + go func() { + // Await the original future's result, respecting the passed + // context for cancellation. + result := f.Await(ctx) + + // Call the callback function with the result. + cFunc(result) + }() +} diff --git a/actor/future_test.go b/actor/future_test.go new file mode 100644 index 000000000..3d56d2c52 --- /dev/null +++ b/actor/future_test.go @@ -0,0 +1,467 @@ +package actor + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" + "pgregory.net/rapid" +) + +// TestFutureAwaitContextCancellation tests that Await respects context +// cancellation if the context is cancelled before the future resolves. +func TestFutureAwaitContextCancellation(t *testing.T) { + t.Parallel() + + rapid.Check(t, func(t *rapid.T) { + // Test cancellation when the Await context is cancelled via + // context.Cancel. The underlying future will not be completed, allowing + // us to test the cancellation path of Await. + prom1 := NewPromise[int]() + fut1 := prom1.Future() + ctx1, cancel1 := context.WithCancel(context.Background()) + + // We'll cancel the future immediately after creating it. + cancel1() + + result1 := fut1.Await(ctx1) + + require.True(t, result1.IsErr()) + require.ErrorIs( + t, result1.Err(), context.Canceled, + "await with immediate cancel", + ) + + // Test cancellation when the Await context times out. The + // underlying future will also not be completed. + prom2 := NewPromise[int]() + fut2 := prom2.Future() + + // Use a very short timeout that will trigger. + ctx2, cancel2 := context.WithTimeout( + context.Background(), 1*time.Nanosecond, + ) + defer cancel2() + + // Await the future; it should fall through to the timeout + // because the future itself is not completed. + result2 := fut2.Await(ctx2) + + require.True(t, result2.IsErr()) + require.ErrorIs( + t, result2.Err(), context.DeadlineExceeded, + "await with timeout", + ) + }) +} + +// TestFutureAwaitFutureCompletes tests that Await returns the future's +// result if the context is not cancelled before the future resolves. +func TestFutureAwaitFutureCompletes(t *testing.T) { + t.Parallel() + + rapid.Check(t, func(t *rapid.T) { + valToSet := rapid.Int().Draw(t, "valToSet") + + // With a 50% chance, configure the test to complete the future + // with an error instead of a successful value. + var errToSet error + if rapid.Bool().Draw(t, "have_error") { + errToSet = fmt.Errorf("err") + } + + promise := NewPromise[int]() + fut := promise.Future() + + // Use a background context for Await, as we expect the future + // to complete normally. + ctx := context.Background() + + // Complete the future in a separate goroutine to simulate an + // asynchronous operation. + go func() { + if errToSet != nil { + promise.Complete(fn.Err[int](errToSet)) + } else { + promise.Complete(fn.Ok(valToSet)) + } + }() + + // Now we'll wait for the future to complete, then verify below + // that the result (value or error) is as expected. + result := fut.Await(ctx) + + if errToSet != nil { + // If an error was set, verify that Await returns that + // specific error. + require.True(t, result.IsErr()) + require.ErrorIs( + t, result.Err(), errToSet, + "await with error", + ) + } else { + // If no error was set, verify that Await returns the + // correct value. + require.False(t, result.IsErr(), "await with value") + + result.WhenOk(func(val int) { + require.Equal( + t, valToSet, val, "await with value", + ) + }) + } + }) +} + +// TestFutureThenApplyContextCancellation tests that ThenApply respects its +// context, yielding a context error if cancelled before the original future +// completes. +func TestFutureThenApplyContextCancellation(t *testing.T) { + t.Parallel() + + rapid.Check(t, func(t *rapid.T) { + // The original future will not be completed in this test case, + // allowing us to specifically test the cancellation behavior of + // the context passed to ThenApply. + originalPromise := NewPromise[int]() + originalFut := originalPromise.Future() + + // Create a context for ThenApply and cancel it immediately. + ctxApply, cancelApply := context.WithCancel( + context.Background(), + ) + cancelApply() + + var transformCalled atomic.Bool + transform := func(i int) int { + transformCalled.Store(true) + return i * 2 + } + + // Register the transformation. The ThenApply operation itself + // will start a goroutine to await the originalFut. + newFut := originalFut.ThenApply(ctxApply, transform) + + // Await the new (transformed) future. Use a background context + // for this Await to isolate the test to the cancellation of + // ctxApply. + result := newFut.Await(context.Background()) + + require.True(t, result.IsErr()) + require.ErrorIs( + t, result.Err(), context.Canceled, + "ThenApply with cancelled context", + ) + require.False( + t, transformCalled.Load(), + "ThenApply transform function called despite "+ + "context cancellation", + ) + }) +} + +// TestFutureThenApplyOriginalFutureCompletes tests ThenApply's behavior when +// the original future completes (with a value or error) before ThenApply's +// context is cancelled. +func TestFutureThenApplyOriginalFutureCompletes(t *testing.T) { + t.Parallel() + + rapid.Check(t, func(t *rapid.T) { + initialVal := rapid.Int().Draw(t, "initialVal") + + // Configure whether the original future completes with an error + // or a successful value. + var originalErr error + if rapid.Bool().Draw(t, "have_error") { + originalErr = fmt.Errorf("original error") + } + + originalPromise := NewPromise[int]() + originalFut := originalPromise.Future() + + // Create a context for ThenApply that should not cancel before + // the original future completes. + ctxApply, cancelApply := context.WithTimeout( + context.Background(), 50*time.Millisecond, + ) + defer cancelApply() + + var transformCalled atomic.Bool + transform := func(i int) int { + transformCalled.Store(true) + return i * 2 + } + + newFut := originalFut.ThenApply(ctxApply, transform) + + // Complete the original future in a separate goroutine to + // simulate asynchrony. + go func() { + if originalErr != nil { + originalPromise.Complete( + fn.Err[int](originalErr), + ) + } else { + originalPromise.Complete(fn.Ok(initialVal)) + } + }() + + // Await our new future which transforms the original future's + // result. Use a background context for this Await. + result := newFut.Await(context.Background()) + + if originalErr != nil { + // If the original future had an error, the transformed + // future should also yield that same error. + require.True(t, result.IsErr()) + require.ErrorIs( + t, result.Err(), originalErr, + "ThenApply with original error", + ) + require.False( + t, transformCalled.Load(), + "ThenApply transform function called despite "+ + "original future having an error", + ) + } else { + // If the original future completed successfully, the + // transformed future should contain the transformed value. + require.False( + t, result.IsErr(), + "ThenApply with original value", + ) + require.True( + t, transformCalled.Load(), + "ThenApply transform function not called for "+ + "successful original future", + ) + + result.WhenOk(func(val int) { + expectedTransformedVal := initialVal * 2 + require.Equal( + t, expectedTransformedVal, val, + "ThenApply with original value", + ) + }) + } + }) +} + +// TestFutureOnCompleteContextCancellation tests that OnComplete's callback +// receives a context error if its context is cancelled before the future +// completes. +func TestFutureOnCompleteContextCancellation(t *testing.T) { + t.Parallel() + + rapid.Check(t, func(t *rapid.T) { + // The original future will not complete in this test, allowing + // us to focus on the cancellation of OnComplete's context. + originalPromise := NewPromise[int]() + originalFut := originalPromise.Future() + + // Create a context for OnComplete and cancel it immediately to + // simulate a premature cancellation. + ctxComplete, cancelComplete := context.WithCancel( + context.Background(), + ) + cancelComplete() + + var wg sync.WaitGroup + wg.Add(1) + var ( + callbackInvoked atomic.Bool + callbackResultValue fn.Result[int] + + // mu is a mutex to protect callbackResultValue as it's + // written by the callback goroutine and read by the + // test goroutine. + mu sync.Mutex + ) + + // Register an OnComplete callback. The callback itself runs in + // a new goroutine started by OnComplete. + originalFut.OnComplete(ctxComplete, func(res fn.Result[int]) { + mu.Lock() + callbackResultValue = res + mu.Unlock() + + callbackInvoked.Store(true) + wg.Done() + }) + + // Use a wait group and a channel to wait for the callback to + // be invoked. + waitChan := make(chan struct{}) + go func() { + wg.Wait() + close(waitChan) + }() + + select { + // The callback should be invoked, even if with a context error. + case <-waitChan: + case <-time.After(50 * time.Millisecond): + require.Fail( + t, "OnComplete callback timed out waiting "+ + "for execution after context cancel", + ) + } + + require.True( + t, callbackInvoked.Load(), + "OnComplete callback not invoked", + ) + + mu.Lock() + defer mu.Unlock() + + // Verify that the callback received a context.Canceled error + // because its context (ctxComplete) was cancelled. + require.True(t, callbackResultValue.IsErr()) + require.ErrorIs( + t, callbackResultValue.Err(), context.Canceled, + "OnComplete with cancelled context", + ) + }) +} + +// TestFutureOnCompleteFutureCompletes tests OnComplete's behavior when the +// future completes (with value or error) before its context is cancelled. +func TestFutureOnCompleteFutureCompletes(t *testing.T) { + t.Parallel() + + rapid.Check(t, func(t *rapid.T) { + valToSet := rapid.Int().Draw(t, "valToSet") + + // Configure whether the original future completes with an error + // or a successful value. + var originalErr error + if rapid.Bool().Draw(t, "have_error") { + originalErr = fmt.Errorf("original error") + } + + originalPromise := NewPromise[int]() + originalFut := originalPromise.Future() + + // Use a background context for OnComplete, as we expect the + // future to complete normally. + ctxComplete := context.Background() + + var wg sync.WaitGroup + wg.Add(1) + + var ( + callbackInvoked atomic.Bool + callbackResultValue fn.Result[int] + mu sync.Mutex + ) + + // Register an OnComplete callback. This callback will execute + // once the originalFut completes. + originalFut.OnComplete(ctxComplete, func(res fn.Result[int]) { + mu.Lock() + callbackResultValue = res + mu.Unlock() + + callbackInvoked.Store(true) + + wg.Done() + }) + + // Complete the original future in a separate goroutine to + // simulate an asynchronous operation. + go func() { + if originalErr != nil { + originalPromise.Complete( + fn.Err[int](originalErr), + ) + } else { + originalPromise.Complete(fn.Ok(valToSet)) + } + }() + + // Use a wait group and a channel to wait for the callback's + // execution. + waitChan := make(chan struct{}) + go func() { + wg.Wait() + close(waitChan) + }() + + select { + // The callback should be invoked as the future completes. + case <-waitChan: + case <-time.After(50 * time.Millisecond): + require.Fail( + t, "OnComplete callback timed out waiting "+ + "for execution", + ) + } + + require.True(t, callbackInvoked.Load()) + + mu.Lock() + defer mu.Unlock() + + // Verify that the callback received the correct result (either + // the error or the value from the completed future). + if originalErr != nil { + require.True(t, callbackResultValue.IsErr()) + require.ErrorIs( + t, callbackResultValue.Err(), originalErr, + "OnComplete with error", + ) + } else { + require.False( + t, callbackResultValue.IsErr(), + "OnComplete with value", + ) + callbackResultValue.WhenOk(func(val int) { + require.Equal( + t, valToSet, val, + "OnComplete with value", + ) + }) + } + }) +} + +// TestPromiseCompleteIdempotency verifies that calling Complete on a Promise +// multiple times is safe and only the first completion takes effect. Subsequent +// calls should return false and not alter the future's result. +func TestPromiseCompleteIdempotency(t *testing.T) { + t.Parallel() + + promise := NewPromise[string]() + future := promise.Future() + + // First completion should succeed. + firstResult := fn.Ok("first-value") + ok := promise.Complete(firstResult) + require.True(t, ok, "first Complete should return true") + + // Second completion with a different value should be ignored. + secondResult := fn.Ok("second-value") + ok = promise.Complete(secondResult) + require.False(t, ok, "second Complete should return false") + + // Third completion with an error should also be ignored. + thirdResult := fn.Err[string](fmt.Errorf("should be ignored")) + ok = promise.Complete(thirdResult) + require.False(t, ok, "third Complete should return false") + + // The future should contain the first value. + result := future.Await(context.Background()) + require.False(t, result.IsErr(), "future should not be an error") + result.WhenOk(func(val string) { + require.Equal( + t, "first-value", val, + "future should contain the first completion value", + ) + }) +} diff --git a/actor/interface.go b/actor/interface.go new file mode 100644 index 000000000..8a59509db --- /dev/null +++ b/actor/interface.go @@ -0,0 +1,46 @@ +package actor + +import ( + "context" + "fmt" + + "github.com/lightningnetwork/lnd/fn/v2" +) + +// Future represents the result of an asynchronous computation. It allows +// consumers to wait for the result (Await), apply transformations upon +// completion (ThenApply), or register a callback to be executed when the +// result is available (OnComplete). +type Future[T any] interface { + // Await blocks until the result is available or the context is + // cancelled, then returns it. + Await(ctx context.Context) fn.Result[T] + + // ThenApply registers a function to transform the result of a future. + // The original future is not modified, a new instance of the future is + // returned. If the passed context is cancelled while waiting for the + // original future to complete, the new future will complete with the + // context's error. + ThenApply(ctx context.Context, fn func(T) T) Future[T] + + // OnComplete registers a function to be called when the result of the + // future is ready. If the passed context is cancelled before the future + // completes, the callback function will be invoked with the context's + // error. + OnComplete(ctx context.Context, fn func(fn.Result[T])) +} + +// Promise is an interface that allows for the completion of an associated +// Future. It provides a way to set the result of an asynchronous operation. +// The producer of an asynchronous result uses a Promise to set the outcome, +// while consumers use the associated Future to retrieve it. +type Promise[T any] interface { + // Future returns the Future interface associated with this Promise. + // Consumers can use this to Await the result or register callbacks. + Future() Future[T] + + // Complete attempts to set the result of the future. It returns true if + // this call successfully set the result (i.e., it was the first to + // complete it), and false if the future had already been completed. + Complete(result fn.Result[T]) bool +} From b055af6420736cbb04f9a2cc245859ee53402335 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:55:48 -0700 Subject: [PATCH 3/6] actor: add fundamental interfaces and concrete Actor impl In this commit, we add the actual Actor implementation. We define a series of types and interfaces, that in concert, describe our actor. An actor has some ID, a reference (used to send messages to it), and also a set of defined messages that it'll accept. An actor can be implemented using a simple function if it's stateless. Otherwise, a struct can implement the Receive method, and handle its internal message passing and state that way. --- actor/actor.go | 324 ++++++++++++++++++++++++++++++++ actor/actor_test.go | 446 ++++++++++++++++++++++++++++++++++++++++++++ actor/func_actor.go | 45 +++++ actor/interface.go | 73 ++++++++ 4 files changed, 888 insertions(+) create mode 100644 actor/actor.go create mode 100644 actor/actor_test.go create mode 100644 actor/func_actor.go diff --git a/actor/actor.go b/actor/actor.go new file mode 100644 index 000000000..373898b87 --- /dev/null +++ b/actor/actor.go @@ -0,0 +1,324 @@ +package actor + +import ( + "context" + "sync" + + "github.com/lightningnetwork/lnd/fn/v2" +) + +// ActorConfig holds the configuration parameters for creating a new Actor. +// It is generic over M (Message type) and R (Response type) to accommodate +// the actor's specific behavior. +type ActorConfig[M Message, R any] struct { + // ID is the unique identifier for the actor. + ID string + + // Behavior defines how the actor responds to messages. + Behavior ActorBehavior[M, R] + + // DLO is a reference to the dead letter office for this actor system. + // If nil, undeliverable messages during shutdown or due to a full + // mailbox (if such logic were added) might be dropped. + DLO ActorRef[Message, any] + + // MailboxSize defines the buffer capacity of the actor's mailbox. + MailboxSize int +} + +// envelope wraps a message with its associated promise. This allows the sender +// of an "ask" message to await a response. If the promise is nil, it +// signifies a "tell" operation (fire-and-forget). +type envelope[M Message, R any] struct { + message M + promise Promise[R] +} + +// Actor represents a concrete actor implementation. It encapsulates a behavior, +// manages its internal state implicitly through that behavior, and processes +// messages from its mailbox sequentially in its own goroutine. +type Actor[M Message, R any] struct { + // id is the unique identifier for the actor. + id string + + // behavior defines how the actor responds to messages. + behavior ActorBehavior[M, R] + + // mailbox is the incoming message queue for the actor. + mailbox chan envelope[M, R] + + // ctx is the context governing the actor's lifecycle. + ctx context.Context + + // cancel is the function to cancel the actor's context. + cancel context.CancelFunc + + // dlo is a reference to the dead letter office for this actor system. + dlo ActorRef[Message, any] + + // startOnce ensures the actor's processing loop is started only once. + startOnce sync.Once + + // stopOnce ensures the actor's processing loop is stopped only once. + stopOnce sync.Once + + // ref is the cached ActorRef for this actor. + ref ActorRef[M, R] +} + +// NewActor creates a new actor instance with the given ID and behavior. +// It initializes the actor's internal structures but does not start its +// message processing goroutine. The Start() method must be called to begin +// processing messages. +func NewActor[M Message, R any](cfg ActorConfig[M, R]) (*Actor[M, R], + error) { + + if cfg.ID == "" { + return nil, ErrEmptyActorID + } + + if cfg.Behavior == nil { + return nil, ErrNilBehavior + } + + ctx, cancel := context.WithCancel(context.Background()) + + // Ensure MailboxSize has a sane default if not specified or zero. A + // capacity of 0 would make the channel unbuffered, which is generally + // not desired for actor mailboxes. + mailboxCapacity := cfg.MailboxSize + if mailboxCapacity <= 0 { + // Default to a small capacity if an invalid one is given. This + // could also come from a global constant. + mailboxCapacity = 1 + } + + actor := &Actor[M, R]{ + id: cfg.ID, + behavior: cfg.Behavior, + mailbox: make(chan envelope[M, R], mailboxCapacity), + ctx: ctx, + cancel: cancel, + dlo: cfg.DLO, + } + + // Create and cache the actor's own reference. + actor.ref = &actorRefImpl[M, R]{ + actor: actor, + } + + return actor, nil +} + +// Start initiates the actor's message processing loop in a new goroutine. This +// method should be called once after the actor is created. +func (a *Actor[M, R]) Start() { + a.startOnce.Do(func() { + log.Infof("Actor %s: starting", a.id) + + go a.process() + }) +} + +// process is the main event loop for the actor. It continuously monitors its +// mailbox for incoming messages and its context for cancellation signals. +func (a *Actor[M, R]) process() { + for { + select { + case env := <-a.mailbox: + result := a.behavior.Receive(a.ctx, env.message) + + // If a promise was provided (i.e., it was an "ask" + // operation), complete the promise with the result from + // the behavior. + if env.promise != nil { + env.promise.Complete(result) + } + + // The actor's context has been cancelled, signaling a stop + // request. Exit the processing loop to terminate the actor's + // goroutine. Before exiting, drain any remaining messages from + // the mailbox. + // + // NOTE: We intentionally do NOT close the mailbox channel here. + // Closing it would create a TOCTOU race with Tell/Ask, which + // check ctx.Err() before sending to the mailbox. Between that + // check and the actual send, the channel could be closed, + // causing a panic. Instead, we drain using a non-blocking + // select loop. Messages sent after this point will remain in + // the buffered channel and be garbage collected. + case <-a.ctx.Done(): + log.Debugf("Actor %s: context cancelled, draining "+ + "mailbox", a.id) + + // Drain any remaining messages from the mailbox + // without closing the channel. + drained := 0 + for { + select { + case env := <-a.mailbox: + drained++ + + // If a DLO is configured, send the + // original message there for auditing + // or potential manual reprocessing. + if a.dlo != nil { + a.dlo.Tell( + context.Background(), + env.message, + ) + } + + // If it was an Ask, complete the + // promise with an error indicating the + // actor terminated. + if env.promise != nil { + env.promise.Complete( + fn.Err[R]( + ErrActorTerminated, + ), + ) + } + + default: + if drained > 0 { + log.Debugf("Actor %s: "+ + "drained %d "+ + "message(s) during "+ + "shutdown", + a.id, drained) + } + + // No more messages in the mailbox. + return + } + } + } + } +} + +// Stop signals the actor to terminate its processing loop and shut down. +// This is achieved by cancelling the actor's internal context. The actor's +// goroutine will exit once it detects the context cancellation. +func (a *Actor[M, R]) Stop() { + a.stopOnce.Do(func() { + log.Infof("Actor %s: stopping", a.id) + + a.cancel() + }) +} + +// actorRefImpl provides a concrete implementation of the ActorRef interface. It +// holds a reference to the target Actor instance, enabling message sending. +type actorRefImpl[M Message, R any] struct { + actor *Actor[M, R] +} + +// Tell sends a message without waiting for a response. If the context is +// cancelled before the message can be sent to the actor's mailbox, the message +// may be dropped. +// +//nolint:ll +func (ref *actorRefImpl[M, R]) Tell(ctx context.Context, msg M) { + // If the actor's own context is already done, don't try to send. + // Route to DLO if available. + if ref.actor.ctx.Err() != nil { + ref.trySendToDLO(msg) + return + } + + select { + // Message successfully enqueued in the actor's mailbox. + case ref.actor.mailbox <- envelope[M, R]{message: msg, promise: nil}: + + // The context for the Tell operation was cancelled before the message + // could be enqueued. The message is dropped. + case <-ctx.Done(): + log.Warnf("Tell to actor %s: message %s dropped "+ + "(caller context cancelled)", ref.actor.id, + msg.MessageType()) + + // The actor itself has been stopped/terminated. + case <-ref.actor.ctx.Done(): + // If the actor is terminated and has a DLO, send the message + // there. Otherwise, it's dropped. + ref.trySendToDLO(msg) + } +} + +// Ask sends a message and returns a Future for the response. The Future will be +// completed with the actor's reply or an error if the operation fails (e.g., +// context cancellation before send). +// +//nolint:ll +func (ref *actorRefImpl[M, R]) Ask(ctx context.Context, msg M) Future[R] { + // Create a new promise that will be fulfilled with the actor's response. + promise := NewPromise[R]() + + // If the actor's own context is already done, complete the promise with + // ErrActorTerminated and return immediately. This is the primary guard + // against trying to send to a stopped actor. + if ref.actor.ctx.Err() != nil { + promise.Complete(fn.Err[R](ErrActorTerminated)) + return promise.Future() + } + + // Check if the context is already done before attempting to send. This + // ensures deterministic behavior and prevents a race where the message + // could be enqueued even though the context was already cancelled. + if ctx.Err() != nil { + promise.Complete(fn.Err[R](ctx.Err())) + return promise.Future() + } + + select { + // Attempt to send the message along with its promise to the actor's + // mailbox. + case ref.actor.mailbox <- envelope[M, R]{message: msg, promise: promise}: + + // The context for the Ask operation was cancelled before the message + // could be enqueued. Complete the promise with the context's error to + // unblock the caller. + case <-ctx.Done(): + promise.Complete(fn.Err[R](ctx.Err())) + + // The actor's context was cancelled (e.g., actor stopped) while this + // Ask operation was attempting to send (e.g., mailbox was full). + case <-ref.actor.ctx.Done(): + promise.Complete(fn.Err[R](ErrActorTerminated)) + } + + // Return the future associated with the promise, allowing the caller to + // await the response. + return promise.Future() +} + +// trySendToDLO attempts to send the message to the actor's DLO if configured. +func (ref *actorRefImpl[M, R]) trySendToDLO(msg M) { + if ref.actor.dlo != nil { + // Use context.Background() for sending to DLO as the + // original context might be done or the operation + // should not be bound by it. + // This Tell to DLO is fire-and-forget. + ref.actor.dlo.Tell(context.Background(), msg) + } +} + +// ID returns the unique identifier for this actor. +func (ref *actorRefImpl[M, R]) ID() string { + return ref.actor.id +} + +// Ref returns an ActorRef for this actor. This allows clients to interact with +// the actor (send messages) without having direct access to the Actor struct +// itself, promoting encapsulation and location transparency. +func (a *Actor[M, R]) Ref() ActorRef[M, R] { + return a.ref +} + +// TellRef returns a TellOnlyRef for this actor. This allows clients to send +// messages to the actor using only the "tell" pattern (fire-and-forget), +// without having access to "ask" capabilities. +func (a *Actor[M, R]) TellRef() TellOnlyRef[M] { + return a.ref +} diff --git a/actor/actor_test.go b/actor/actor_test.go new file mode 100644 index 000000000..3d49aa6eb --- /dev/null +++ b/actor/actor_test.go @@ -0,0 +1,446 @@ +package actor + +import ( + "context" + "errors" + "fmt" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// testMsg is a simple message type for testing. It embeds BaseMessage to +// satisfy the actor.Message interface. +type testMsg struct { + BaseMessage + data string + + replyChan chan string +} + +// MessageType returns the type name of the message. +func (m *testMsg) MessageType() string { + return "testMsg" +} + +// newTestMsg creates a new test message. +func newTestMsg(data string) *testMsg { + return &testMsg{data: data} +} + +// newTestMsgWithReply creates a new test message that includes a reply channel. +// This can be used by test behaviors to send data back to the test +// synchronously, especially for Tell operations. +func newTestMsgWithReply(data string, replyChan chan string) *testMsg { + return &testMsg{data: data, replyChan: replyChan} +} + +// echoBehavior is a simple actor behavior that processes *testMsg messages. It +// stores the last message's data and, for Ask, echoes it back. For Tell, if +// replyChan is set in testMsg, it sends data back on it. +type echoBehavior struct { + lastMsgData atomic.Value + processingDelay time.Duration + t *testing.T +} + +// newEchoBehavior creates a new echoBehavior. +func newEchoBehavior(t *testing.T, delay time.Duration) *echoBehavior { + return &echoBehavior{t: t, processingDelay: delay} +} + +// Receive handles incoming messages. It simulates work if processingDelay is +// set, stores the message data, and responds for Ask operations or via +// replyChan for Tell. +func (b *echoBehavior) Receive(_ context.Context, + msg *testMsg) fn.Result[string] { + + if b.processingDelay > 0 { + time.Sleep(b.processingDelay) + } + + b.lastMsgData.Store(msg.data) + + if msg.replyChan != nil { + // Attempt to send the data on the reply channel, but quit if + // it takes longer than 1 second (e.g., channel unbuffered + // and no receiver). + select { + case msg.replyChan <- msg.data: + case <-time.After(time.Second): + b.t.Logf("warning: replyChan send timed out") + } + } + + return fn.Ok(fmt.Sprintf("echo: %s", msg.data)) +} + +// GetLastMsgData retrieves the data from the last message processed. +func (b *echoBehavior) GetLastMsgData() (string, bool) { + val := b.lastMsgData.Load() + if val == nil { + return "", false + } + data, ok := val.(string) + return data, ok +} + +// errorBehavior is an actor behavior that always returns a predefined error +// upon receiving a message. +type errorBehavior struct { + err error +} + +// newErrorBehavior creates a new errorBehavior. +func newErrorBehavior(err error) *errorBehavior { + return &errorBehavior{err: err} +} + +// Receive always returns the configured error. +func (b *errorBehavior) Receive(_ context.Context, + _ *testMsg) fn.Result[string] { + + return fn.Err[string](b.err) +} + +// blockingBehavior is an actor behavior that blocks until its actorCtx is done. +type blockingBehavior struct{} + +// Receive blocks until the actor's context is cancelled, then returns the +// context's error. +func (b *blockingBehavior) Receive(actorCtx context.Context, + _ *testMsg) fn.Result[string] { + + <-actorCtx.Done() + return fn.Err[string](actorCtx.Err()) +} + +// deadLetterTestMsg is a distinct message type used for testing DLO +// interactions. +type deadLetterTestMsg struct { + BaseMessage + id string +} + +// MessageType returns the type name of the message. +func (m *deadLetterTestMsg) MessageType() string { + return "deadLetterTestMsg" +} + +// deadLetterObserverBehavior is a behavior for a test Dead Letter Office actor. +// It records all messages sent to it, allowing tests to verify DLO +// interactions. +type deadLetterObserverBehavior struct { + mu sync.Mutex + receivedMsgs []Message +} + +// newDeadLetterObserverBehavior creates a new deadLetterObserverBehavior. +func newDeadLetterObserverBehavior() *deadLetterObserverBehavior { + return &deadLetterObserverBehavior{ + receivedMsgs: make([]Message, 0), + } +} + +// Receive records the incoming message and returns a successful result. +func (b *deadLetterObserverBehavior) Receive(_ context.Context, + msg Message) fn.Result[any] { + + b.mu.Lock() + b.receivedMsgs = append(b.receivedMsgs, msg) + b.mu.Unlock() + + return fn.Ok[any](nil) +} + +// GetReceivedMsgs returns a copy of all messages received by this DLO. +func (b *deadLetterObserverBehavior) GetReceivedMsgs() []Message { + b.mu.Lock() + defer b.mu.Unlock() + + msgs := make([]Message, len(b.receivedMsgs)) + copy(msgs, b.receivedMsgs) + + return msgs +} + +// actorTestHarness provides helper methods for setting up actors in tests. It +// manages a dedicated DLO for actors created through it. +type actorTestHarness struct { + t *testing.T + dlo *Actor[Message, any] + dloBeh *deadLetterObserverBehavior +} + +// newActorTestHarness sets up a test harness with a dedicated DLO. The DLO is +// automatically stopped when the test cleans up. +func newActorTestHarness(t *testing.T) *actorTestHarness { + t.Helper() + + dloBeh := newDeadLetterObserverBehavior() + dloCfg := ActorConfig[Message, any]{ + ID: "test-dlo-" + t.Name(), + Behavior: dloBeh, + DLO: nil, + MailboxSize: 10, + } + dloActor, err := NewActor[Message, any](dloCfg) + require.NoError(t, err) + dloActor.Start() + + t.Cleanup(dloActor.Stop) + + return &actorTestHarness{ + t: t, + dlo: dloActor, + dloBeh: dloBeh, + } +} + +// newActor creates, starts, and registers a new actor for cleanup. The actor +// will use the harness's DLO. +func (h *actorTestHarness) newActor(id string, + beh ActorBehavior[*testMsg, string], + mailboxSize int) *Actor[*testMsg, string] { + + h.t.Helper() + + cfg := ActorConfig[*testMsg, string]{ + ID: id, + Behavior: beh, + DLO: h.dlo.Ref(), + MailboxSize: mailboxSize, + } + actor, err := NewActor(cfg) + require.NoError(h.t, err) + actor.Start() + + h.t.Cleanup(actor.Stop) + + return actor +} + +// assertDLOMessage checks that the DLO eventually receives a specific message. +func (h *actorTestHarness) assertDLOMessage(expectedMsg Message) { + h.t.Helper() + require.Eventually(h.t, func() bool { + msgs := h.dloBeh.GetReceivedMsgs() + for _, m := range msgs { + if reflect.DeepEqual(m, expectedMsg) { + return true + } + } + return false + }, time.Second, 10*time.Millisecond, + "dLO did not receive expected message: %v", expectedMsg, + ) +} + +// assertNoDLOMessages checks that the DLO has not received any messages. +func (h *actorTestHarness) assertNoDLOMessages() { + h.t.Helper() + + // Allow a very brief moment for any async DLO sends to occur. + time.Sleep(20 * time.Millisecond) + + msgs := h.dloBeh.GetReceivedMsgs() + + require.Empty(h.t, msgs, "dLO received unexpected messages") +} + +// TestActorNewActorIDAndRefs verifies that NewActor correctly initializes an +// actor's ID and provides functional ActorRef and TellOnlyRef instances. +func TestActorNewActorIDAndRefs(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + actorID := "test-actor-1" + beh := newEchoBehavior(t, 0) + actor := h.newActor(actorID, beh, 1) + + require.Equal(t, actorID, actor.Ref().ID(), "actorRef ID mismatch") + require.Equal( + t, actorID, actor.TellRef().ID(), "tellOnlyRef ID mismatch", + ) + require.NotNil(t, actor.Ref(), "actorRef should not be nil") + require.NotNil(t, actor.TellRef(), "tellOnlyRef should not be nil") +} + +// TestActorStartStop verifies the basic lifecycle of an actor: starting, +// processing messages, and stopping. +func TestActorStartStop(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + beh := newEchoBehavior(t, 0) + actor := h.newActor("test-actor-lifecycle", beh, 1) + + // Actor should be running and process a message. + msgData := "hello" + replyChan := make(chan string, 1) + actor.Ref().Tell( + context.Background(), newTestMsgWithReply(msgData, replyChan), + ) + + received, err := fn.RecvOrTimeout(replyChan, 100*time.Millisecond) + require.NoError(t, err, "timed out waiting for actor to process message") + require.Equal( + t, msgData, received, "actor did not process message before stop", + ) + + actor.Stop() + time.Sleep(50 * time.Millisecond) + + // Try sending another message; it should ideally not be processed or go + // to DLO. + msgDataAfterStop := "message-after-stop" + replyChanAfterStop := make(chan string, 1) + actor.Ref().Tell( + context.Background(), + newTestMsgWithReply(msgDataAfterStop, replyChanAfterStop), + ) + + // We expect a timeout here, meaning the message was not processed by + // the echoBehavior's replyChan. + _, err = fn.RecvOrTimeout(replyChanAfterStop, 100*time.Millisecond) + // err == nil would mean a message was received, meaning the actor + // processed it after Stop(). + require.Error(t, err, "actor processed message after Stop()") + require.ErrorContains(t, err, "timeout hit") + + h.assertDLOMessage( + &testMsg{data: msgDataAfterStop, replyChan: replyChanAfterStop}, + ) +} + +// TestActorTellBasic verifies that a message sent via Tell is processed by the +// actor's behavior. +func TestActorTellBasic(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + beh := newEchoBehavior(t, 0) + actor := h.newActor("test-actor-tell", beh, 1) + + msgData := "tell-message" + replyChan := make(chan string, 1) + actor.Ref().Tell( + context.Background(), newTestMsgWithReply(msgData, replyChan), + ) + + receivedTell, errTell := fn.RecvOrTimeout(replyChan, 100*time.Millisecond) + require.NoError(t, errTell, "timed out waiting for Tell message processing") + require.Equal( + t, msgData, receivedTell, "behavior did not receive Tell message data", + ) + + lastData, ok := beh.GetLastMsgData() + require.True(t, ok, "last message data not set in behavior") + require.Equal(t, msgData, lastData, "last message data mismatch") + h.assertNoDLOMessages() +} + +// TestActorAskSuccess verifies that a message sent via Ask is processed, and +// the returned Future is completed with the behavior's successful result. +func TestActorAskSuccess(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + beh := newEchoBehavior(t, 0) + actor := h.newActor("test-actor-ask-success", beh, 1) + + msgData := "ask-message" + future := actor.Ref().Ask(context.Background(), newTestMsg(msgData)) + + result := future.Await(context.Background()) + require.False(t, result.IsErr(), "ask returned an error: %v", result.Err()) + + result.WhenOk(func(val string) { + expectedReply := fmt.Sprintf("echo: %s", msgData) + require.Equal(t, expectedReply, val, "ask response mismatch") + }) + + lastData, ok := beh.GetLastMsgData() + require.True(t, ok, "last message data not set in behavior") + require.Equal(t, msgData, lastData, "last message data mismatch") + h.assertNoDLOMessages() +} + +// TestActorAskErrorBehavior verifies that if an actor's behavior returns an +// error, the Future from an Ask call is completed with that error. +func TestActorAskErrorBehavior(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + expectedErr := errors.New("behavior error") + beh := newErrorBehavior(expectedErr) + actor := h.newActor("test-actor-ask-error", beh, 1) + + future := actor.Ref().Ask( + context.Background(), newTestMsg("ask-error-test"), + ) + + result := future.Await(context.Background()) + require.True(t, result.IsErr(), "ask should have returned an error") + require.ErrorIs(t, result.Err(), expectedErr, "ask error mismatch") + + h.assertNoDLOMessages() +} + +// TestFunctionBehaviorFromSimple verifies that FunctionBehaviorFromSimple +// correctly adapts a simple (msg) -> (result, error) function into an +// ActorBehavior, handling both success and error cases. +func TestFunctionBehaviorFromSimple(t *testing.T) { + t.Parallel() + + t.Run("success", func(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + + beh := FunctionBehaviorFromSimple( + func(msg *testMsg) (string, error) { + return "simple: " + msg.data, nil + }, + ) + actor := h.newActor("test-simple-success", beh, 1) + + future := actor.Ref().Ask( + context.Background(), newTestMsg("hello"), + ) + result := future.Await(context.Background()) + require.False( + t, result.IsErr(), + "expected success, got: %v", result.Err(), + ) + result.WhenOk(func(val string) { + require.Equal(t, "simple: hello", val) + }) + }) + + t.Run("error", func(t *testing.T) { + t.Parallel() + + h := newActorTestHarness(t) + + expectedErr := errors.New("simple behavior error") + beh := FunctionBehaviorFromSimple( + func(msg *testMsg) (string, error) { + return "", expectedErr + }, + ) + actor := h.newActor("test-simple-error", beh, 1) + + future := actor.Ref().Ask( + context.Background(), newTestMsg("hello"), + ) + result := future.Await(context.Background()) + require.True(t, result.IsErr()) + require.ErrorIs(t, result.Err(), expectedErr) + }) +} diff --git a/actor/func_actor.go b/actor/func_actor.go new file mode 100644 index 000000000..f1580f03f --- /dev/null +++ b/actor/func_actor.go @@ -0,0 +1,45 @@ +package actor + +import ( + "context" + + "github.com/lightningnetwork/lnd/fn/v2" +) + +// ActorFunc is a function type that represents an actor which functions purely +// based on a simple function processor. +type ActorFunc[M Message, R any] func(context.Context, M) fn.Result[R] + +// FunctionBehavior adapts a function to the ActorBehavior interface. +type FunctionBehavior[M Message, R any] struct { + fn ActorFunc[M, R] +} + +// NewFunctionBehavior creates a behavior from a function. +func NewFunctionBehavior[M Message, R any]( + fn ActorFunc[M, R]) *FunctionBehavior[M, R] { + + return &FunctionBehavior[M, R]{fn: fn} +} + +// Receive implements ActorBehavior interface for the function. +// +// TODO(roasbeef): just base it off the function direct instead? +func (b *FunctionBehavior[M, R]) Receive(ctx context.Context, + msg M) fn.Result[R] { + + return b.fn(ctx, msg) +} + +// FunctionBehaviorFromSimple adapts a simpler function to the ActorBehavior +// interface. +func FunctionBehaviorFromSimple[M Message, R any]( + sFunc func(M) (R, error)) *FunctionBehavior[M, R] { + + return NewFunctionBehavior( + func(ctx context.Context, msg M) fn.Result[R] { + val, err := sFunc(msg) + return fn.NewResult(val, err) + }, + ) +} diff --git a/actor/interface.go b/actor/interface.go index 8a59509db..6ddbe8fcc 100644 --- a/actor/interface.go +++ b/actor/interface.go @@ -7,6 +7,44 @@ import ( "github.com/lightningnetwork/lnd/fn/v2" ) +// ErrActorTerminated indicates that an operation failed because the target +// actor was terminated or in the process of shutting down. +var ErrActorTerminated = fmt.Errorf("actor terminated") + +// ErrEmptyActorID is returned when an actor is created with an empty ID. +var ErrEmptyActorID = fmt.Errorf("actor ID must not be empty") + +// ErrNilBehavior is returned when an actor is created with a nil behavior. +var ErrNilBehavior = fmt.Errorf("actor behavior must not be nil") + +// ErrDuplicateActorID is returned when attempting to register an actor with an +// ID that is already in use within the actor system. +var ErrDuplicateActorID = fmt.Errorf("actor ID already registered") + +// BaseMessage is a helper struct that can be embedded in message types defined +// outside the actor package to satisfy the Message interface's unexported +// messageMarker method. +type BaseMessage struct{} + +// messageMarker implements the unexported method for the Message interface, +// allowing types that embed BaseMessage to satisfy the Message interface. +func (BaseMessage) messageMarker() {} + +// Message is a sealed interface for actor messages. Actors will receive +// messages conforming to this interface. The interface is "sealed" by the +// unexported messageMarker method, meaning only types that can satisfy it +// (e.g., by embedding BaseMessage or being in the same package) can be +// Messages. +type Message interface { + // messageMarker is a private method that makes this a sealed interface + // (see BaseMessage for embedding). + messageMarker() + + // MessageType returns the type name of the message for + // routing/filtering. + MessageType() string +} + // Future represents the result of an asynchronous computation. It allows // consumers to wait for the result (Await), apply transformations upon // completion (ThenApply), or register a callback to be executed when the @@ -44,3 +82,38 @@ type Promise[T any] interface { // complete it), and false if the future had already been completed. Complete(result fn.Result[T]) bool } + +// TellOnlyRef is a reference to an actor that only supports "tell" operations. +// This is useful for scenarios where only fire-and-forget message passing is +// needed, or to restrict capabilities. +type TellOnlyRef[M Message] interface { + // Tell sends a message without waiting for a response. If the + // context is cancelled before the message can be sent to the actor's + // mailbox, the message may be dropped. + Tell(ctx context.Context, msg M) + + // ID returns the unique identifier for this actor. + ID() string +} + +// ActorRef is a reference to an actor that supports both "tell" and "ask" +// operations. It embeds TellOnlyRef and adds the Ask method for +// request-response interactions. +type ActorRef[M Message, R any] interface { + TellOnlyRef[M] + + // Ask sends a message and returns a Future for the response. + // The Future will be completed with the actor's reply or an error + // if the operation fails (e.g., context cancellation before send). + Ask(ctx context.Context, msg M) Future[R] +} + +// ActorBehavior defines the logic for how an actor processes incoming messages. +// It is a strategy interface that encapsulates the actor's reaction to +// messages. +type ActorBehavior[M Message, R any] interface { + // Receive processes a message and returns a Result. The provided + // context is the actor's internal context, which can be used to + // detect actor shutdown requests. + Receive(actorCtx context.Context, msg M) fn.Result[R] +} From 9ebd8d240dcf57eba084af89adc539542f9e2f26 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:57:48 -0700 Subject: [PATCH 4/6] actor: add the actor system and router In this commit, we add the actor system (along with the receiptionist) and the router. An actor can be registered with the system, which allows other callers to locate it to send message to it via the receptionist. Custom routers can be created for when there're actors that rely on the same service key and also req+resp type. This can be used to implement something similar to a worker pool. --- actor/router.go | 154 +++++++ actor/system.go | 421 +++++++++++++++++++ actor/system_test.go | 958 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1533 insertions(+) create mode 100644 actor/router.go create mode 100644 actor/system.go create mode 100644 actor/system_test.go diff --git a/actor/router.go b/actor/router.go new file mode 100644 index 000000000..87e2dae44 --- /dev/null +++ b/actor/router.go @@ -0,0 +1,154 @@ +package actor + +import ( + "context" + "errors" + "sync/atomic" + + "github.com/lightningnetwork/lnd/fn/v2" +) + +// ErrNoActorsAvailable is returned when a router cannot find any actors +// registered for its service key to forward a message to. +var ErrNoActorsAvailable = errors.New("no actors available for service key") + +// Compile-time assertion that Router satisfies the ActorRef interface. +var _ ActorRef[Message, any] = (*Router[Message, any])(nil) + +// RoutingStrategy defines the interface for selecting an actor from a list of +// available actors. +// The M (Message) and R (Response) type parameters ensure that the strategy +// is compatible with the types of actors it will be selecting. +type RoutingStrategy[M Message, R any] interface { + // Select chooses an ActorRef from the provided slice. It returns the + // selected actor or an error if no actor can be selected (e.g., if the + // list is empty or another strategy-specific issue occurs). + Select(refs []ActorRef[M, R]) (ActorRef[M, R], error) +} + +// RoundRobinStrategy implements a round-robin selection strategy. It is generic +// over M and R to match the RoutingStrategy interface, though its logic doesn't +// depend on these types directly for the selection mechanism itself. +type RoundRobinStrategy[M Message, R any] struct { + // index is used to pick the next actor in a round-robin fashion. It + // must be accessed atomically to ensure thread-safety if multiple + // goroutines use the same strategy instance (which they will via the + // router). + index uint64 +} + +// NewRoundRobinStrategy creates a new RoundRobinStrategy, initialized for +// round-robin selection. +func NewRoundRobinStrategy[M Message, R any]() *RoundRobinStrategy[M, R] { + return &RoundRobinStrategy[M, R]{} +} + +// Select picks an actor from the list using a round-robin algorithm. +func (s *RoundRobinStrategy[M, R]) Select( + refs []ActorRef[M, R], +) (ActorRef[M, R], error) { + if len(refs) == 0 { + return nil, ErrNoActorsAvailable + } + + // Atomically increment and get the current index for selection. + // We subtract 1 because AddUint64 returns the new value (which is + // 1-based for the first call after initialization to 0), and slice + // indexing is 0-based. + idx := atomic.AddUint64(&s.index, 1) - 1 + selectedRef := refs[idx%uint64(len(refs))] + + return selectedRef, nil +} + +// Router is a message-dispatching component that fronts multiple actors +// registered under a specific ServiceKey. It uses a RoutingStrategy to +// distribute messages to one of the available actors. It is generic over M +// (Message type) and R (Response type) to match the actors it routes to. +type Router[M Message, R any] struct { + receptionist *Receptionist + serviceKey ServiceKey[M, R] + strategy RoutingStrategy[M, R] + dlo ActorRef[Message, any] // Dead Letter Office reference. +} + +// NewRouter creates a new Router for a given service key and strategy. The +// receptionist is used to discover actors registered with the service key. +// The router itself is not an actor but a message dispatcher that behaves like +// an ActorRef from the sender's perspective. +func NewRouter[M Message, R any](receptionist *Receptionist, + key ServiceKey[M, R], strategy RoutingStrategy[M, R], + dlo ActorRef[Message, any]) *Router[M, R] { + + return &Router[M, R]{ + receptionist: receptionist, + serviceKey: key, + strategy: strategy, + dlo: dlo, + } +} + +// getActor dynamically finds available actors for the service key and selects +// one using the configured strategy. This method is called internally by Tell +// and Ask on each invocation to ensure up-to-date actor discovery. +func (r *Router[M, R]) getActor() (ActorRef[M, R], error) { + // Discover available actors from the receptionist. + availableActors := FindInReceptionist(r.receptionist, r.serviceKey) + if len(availableActors) == 0 { + return nil, ErrNoActorsAvailable + } + + // Select one actor using the strategy. + return r.strategy.Select(availableActors) +} + +// Tell sends a message to one of the actors managed by the router, selected by +// the routing strategy. If no actors are available or the send context is +// cancelled before the message can be enqueued in the target actor's mailbox, +// the message may be dropped. Errors during actor selection (e.g., +// ErrNoActorsAvailable) are currently not propagated from Tell, aligning with +// its fire-and-forget nature. Such errors could be logged internally if needed. +func (r *Router[M, R]) Tell(ctx context.Context, msg M) { + selectedActor, err := r.getActor() + if err != nil { + // If no actors are available for the service, and a DLO is + // configured, forward the message there. + if errors.Is(err, ErrNoActorsAvailable) && r.dlo != nil { + r.dlo.Tell(context.Background(), msg) + } else { + log.Warnf("Router(%s): message %s dropped "+ + "(no actors available, no DLO configured)", + r.serviceKey.name, msg.MessageType()) + } + + return + } + + selectedActor.Tell(ctx, msg) +} + +// Ask sends a message to one of the actors managed by the router, selected by +// the routing strategy, and returns a Future for the response. If no actors are +// available (ErrNoActorsAvailable), the Future will be completed with this +// error. If the send context is cancelled before the message can be enqueued in +// the chosen actor's mailbox, the Future will be completed with the context's +// error. +func (r *Router[M, R]) Ask(ctx context.Context, msg M) Future[R] { + selectedActor, err := r.getActor() + if err != nil { + // If no actor could be selected (e.g., none available), + // complete the promise immediately with the selection error. + promise := NewPromise[R]() + promise.Complete(fn.Err[R](err)) + return promise.Future() + } + + return selectedActor.Ask(ctx, msg) +} + +// ID provides an identifier for the router. Since a router isn't an actor +// itself but a dispatcher for a service, its ID can be based on the service +// key. +func (r *Router[M, R]) ID() string { + return "router(" + r.serviceKey.name + ")" +} diff --git a/actor/system.go b/actor/system.go new file mode 100644 index 000000000..5e765de6e --- /dev/null +++ b/actor/system.go @@ -0,0 +1,421 @@ +package actor + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/lightningnetwork/lnd/fn/v2" +) + +// stoppable defines an interface for components that can be stopped. +// This is unexported as it's an internal detail of ActorSystem for managing +// actors that need to be shut down. +type stoppable interface { + Stop() +} + +// SystemConfig holds configuration parameters for the ActorSystem. +type SystemConfig struct { + // MailboxCapacity is the default capacity for actor mailboxes. + MailboxCapacity int +} + +// DefaultConfig returns a default configuration for the ActorSystem. +// The default mailbox capacity of 100 means each actor can buffer up to 100 +// pending messages (envelopes). Each envelope holds a message and an optional +// promise pointer, so the memory overhead per actor is roughly proportional to +// the size of the messages being sent multiplied by this capacity. +func DefaultConfig() SystemConfig { + return SystemConfig{ + MailboxCapacity: 100, + } +} + +// ActorSystem manages the lifecycle of actors and provides coordination +// services such as a receptionist for actor discovery and a dead letter office +// for undeliverable messages. It also handles the graceful shutdown of all +// managed actors. +type ActorSystem struct { + // receptionist is used for actor discovery. + receptionist *Receptionist + + // actors stores all actors managed by the system, keyed by their ID. + // This includes the deadLetterActor. + actors map[string]stoppable + + // deadLetterActor handles undeliverable messages. + deadLetterActor ActorRef[Message, any] + + // config holds the system-wide configuration. + config SystemConfig + + // mu protects the 'actors' map. + mu sync.RWMutex + + // ctx is the main context for the actor system. + ctx context.Context + + // cancel cancels the main system context. + cancel context.CancelFunc +} + +// NewActorSystem creates a new actor system using the default configuration. +func NewActorSystem() *ActorSystem { + return NewActorSystemWithConfig(DefaultConfig()) +} + +// NewActorSystemWithConfig creates a new actor system with custom configuration +func NewActorSystemWithConfig(config SystemConfig) *ActorSystem { + ctx, cancel := context.WithCancel(context.Background()) + + // Initialize the core ActorSystem components. + system := &ActorSystem{ + receptionist: newReceptionist(), + config: config, + actors: make(map[string]stoppable), + ctx: ctx, + cancel: cancel, + } + + // Define the behavior for the dead letter actor. It logs undeliverable + // messages and returns an error. + deadLetterBehavior := NewFunctionBehavior( + func(ctx context.Context, msg Message) fn.Result[any] { + log.Warnf("Dead letter received: message type=%s", + msg.MessageType()) + + return fn.Err[any](errors.New( + "message undeliverable: " + msg.MessageType(), + )) + }, + ) + + // Create the raw dead letter actor (*Actor instance). The DLO's own DLO + // reference is nil to prevent loops if messages to the DLO itself fail. + deadLetterActorCfg := ActorConfig[Message, any]{ + ID: "dead-letters", + Behavior: deadLetterBehavior, + DLO: nil, + MailboxSize: config.MailboxCapacity, + } + deadLetterRawActor, err := NewActor[Message, any](deadLetterActorCfg) + if err != nil { + // This should never happen since we control the DLO config. + panic("failed to create dead letter actor: " + err.Error()) + } + deadLetterRawActor.Start() + system.deadLetterActor = deadLetterRawActor.Ref() + + // Add the raw actor to the map of stoppable actors. No lock needed here + // as 'system' is not yet accessible concurrently. + system.actors[deadLetterRawActor.id] = deadLetterRawActor + + // The system is now fully initialized and ready. + return system +} + +// RegisterWithSystem creates an actor with the given ID, service key, and +// behavior within the specified ActorSystem. It starts the actor, adds it to +// the system's management, registers it with the receptionist using the +// provided key, and returns its ActorRef. +func RegisterWithSystem[M Message, R any](as *ActorSystem, id string, + key ServiceKey[M, R], + behavior ActorBehavior[M, R]) (ActorRef[M, R], error) { + + actorCfg := ActorConfig[M, R]{ + ID: id, + Behavior: behavior, + DLO: as.deadLetterActor, + MailboxSize: as.config.MailboxCapacity, + } + // Check for duplicate actor ID before creating the actor. + as.mu.Lock() + if _, exists := as.actors[id]; exists { + as.mu.Unlock() + + return nil, fmt.Errorf("%w: %s", ErrDuplicateActorID, id) + } + + actorInstance, err := NewActor(actorCfg) + if err != nil { + as.mu.Unlock() + + return nil, err + } + actorInstance.Start() + + // Add the actor instance to the system's list of stoppable actors. + as.actors[actorInstance.id] = actorInstance + as.mu.Unlock() + + log.Infof("ActorSystem: registered actor %s with service key %s", + id, key.name) + + // Register the actor's reference with the receptionist under the given + // service key, making it discoverable by other parts of the system. + RegisterWithReceptionist(as.receptionist, key, actorInstance.Ref()) + + return actorInstance.Ref(), nil +} + +// Receptionist returns the system's receptionist, which can be used for +// actor service discovery (finding actors by ServiceKey). +func (as *ActorSystem) Receptionist() *Receptionist { + return as.receptionist +} + +// DeadLetters returns a reference to the system's dead letter actor. Messages +// that cannot be delivered to their intended recipient (e.g., if an Ask +// context is cancelled before enqueuing) may be routed here if not otherwise +// handled. +func (as *ActorSystem) DeadLetters() ActorRef[Message, any] { + return as.deadLetterActor +} + +// Shutdown gracefully stops the actor system. It iterates through all managed +// actors, including the dead letter actor, and calls their Stop method. +// After initiating the stop for all actors, it cancels the main system context. +// This method is safe for concurrent use. +func (as *ActorSystem) Shutdown() error { + log.Infof("ActorSystem: initiating shutdown") + + // Create a slice of actors to stop. This avoids holding the lock while + // calling Stop() on each actor, and includes the dead letter actor. + var actorsToStop []stoppable + as.mu.RLock() + for _, actor := range as.actors { + actorsToStop = append(actorsToStop, actor) + } + as.mu.RUnlock() + + // Notify all managed actors to stop. Actor.Stop() is non-blocking. + // Each actor's Stop method will cancel its internal context, leading + // to the termination of its processing goroutine. + for _, actor := range actorsToStop { + actor.Stop() + } + + // Clear the actors map after initiating their shutdown. + as.mu.Lock() + as.actors = nil + as.mu.Unlock() + + // Finally cancel the main context + // This signals to any other components observing the system's context + // that shutdown has been initiated. + as.cancel() + + return nil +} + +// StopAndRemoveActor stops a specific actor by its ID and removes it from the +// ActorSystem's management. It returns true if the actor was found and stopped, +// false otherwise. +func (as *ActorSystem) StopAndRemoveActor(id string) bool { + as.mu.Lock() + defer as.mu.Unlock() + + actorToStop, exists := as.actors[id] + if !exists { + return false + } + + // Stop the actor. This is non-blocking. + actorToStop.Stop() + + // Remove from the system's management. + delete(as.actors, id) + + return true +} + +// UnregisterFromReceptionist removes an actor reference from a service key in +// the given receptionist. It returns true if the reference was found and +// removed, and false otherwise. This is a package-level generic function +// because methods cannot have their own type parameters in Go. +func UnregisterFromReceptionist[M Message, R any](r *Receptionist, + key ServiceKey[M, R], refToRemove ActorRef[M, R]) bool { + + r.mu.Lock() + defer r.mu.Unlock() + + refs, exists := r.registrations[key.name] + if !exists { + return false + } + + found := false + + // Build a new slice containing only the references that are not the one + // to be removed. + newRefs := make([]any, 0, max(0, len(refs)-1)) + for _, itemInSlice := range refs { + // Try to assert the item from the slice to the specific + // ActorRef[M,R] type we are trying to remove. + if specificActorRef, ok := itemInSlice.(ActorRef[M, R]); ok { + // If the type assertion is successful and it's the one + // we want to remove, mark as found and skip adding it + // to newRefs. + if specificActorRef == refToRemove { + found = true + continue + } + } + newRefs = append(newRefs, itemInSlice) + } + + if !found { + return false + } + + // If the new list of references is empty, remove the key from the map. + // Otherwise, update the map with the new slice. + if len(newRefs) == 0 { + delete(r.registrations, key.name) + } else { + r.registrations[key.name] = newRefs + } + + return true +} + +// ServiceKey is a type-safe identifier used for registering and discovering +// actors via the Receptionist. The generic type parameters M (Message) and R +// (Response) ensure that only actors handling compatible message/response types +// are associated with and retrieved for this key. +type ServiceKey[M Message, R any] struct { + name string +} + +// NewServiceKey creates a new service key with the given name. The name is used +// as the lookup key within the Receptionist. +func NewServiceKey[M Message, R any](name string) ServiceKey[M, R] { + return ServiceKey[M, R]{name: name} +} + +// Spawn registers an actor for this service key within the given ActorSystem. +// It's a convenience method that calls RegisterWithSystem, starting the actor +// and registering it with the receptionist. +func (sk ServiceKey[M, R]) Spawn(as *ActorSystem, id string, + behavior ActorBehavior[M, R]) (ActorRef[M, R], error) { + + return RegisterWithSystem(as, id, sk, behavior) +} + +// Unregister removes an actor reference associated with this service key from +// the ActorSystem's receptionist and also stops the actor. +// It returns true if the actor was successfully unregistered from the +// receptionist AND successfully stopped and removed from the system's +// management. Otherwise, it returns false. +func (sk ServiceKey[M, R]) Unregister(as *ActorSystem, + refToRemove ActorRef[M, R]) bool { + + unregisteredFromReceptionist := UnregisterFromReceptionist( + as.Receptionist(), sk, refToRemove, + ) + + // If not found in receptionist, no need to try stopping. + if !unregisteredFromReceptionist { + return false + } + + // Attempt to stop and remove the actor from the system. + stoppedAndRemoved := as.StopAndRemoveActor(refToRemove.ID()) + + return unregisteredFromReceptionist && stoppedAndRemoved +} + +// UnregisterAll finds all actor references associated with this service key in +// the ActorSystem's receptionist. For each found actor, it attempts to stop it +// and remove it from system management, and also unregisters it from the +// receptionist. +func (sk ServiceKey[M, R]) UnregisterAll(as *ActorSystem) int { + // First find all the refs that match this service key. + refsFound := FindInReceptionist(as.Receptionist(), sk) + + actorsStoppedCount := 0 + for _, ref := range refsFound { + // Attempt to stop and remove the actor from the system's active + // management. This is the primary action to deactivate the + // actor. If StopAndRemoveActor returns true, it means an active + // actor was found in the system's `actors` map and was stopped. + if as.StopAndRemoveActor(ref.ID()) { + actorsStoppedCount++ + } + + // Regardless of whether the actor was actively managed by the + // system (i.e., found in as.actors), attempt to unregister its + // reference from the receptionist. This helps clean up any + // potentially stale entries in the receptionist if an actor was + // removed from the system's management without also being + // unregistered from the receptionist. + UnregisterFromReceptionist(as.Receptionist(), sk, ref) + } + + return actorsStoppedCount +} + +// Receptionist provides service discovery for actors. Actors can be registered +// under a ServiceKey and later discovered by other actors or system components. +type Receptionist struct { + // registrations stores ActorRef instances, keyed by ServiceKey.name. + registrations map[string][]any + + // mu protects access to registrations. + mu sync.RWMutex +} + +// newReceptionist creates a new Receptionist instance. +func newReceptionist() *Receptionist { + return &Receptionist{ + registrations: make(map[string][]any), + } +} + +// RegisterWithReceptionist registers an actor with a service key in the given +// receptionist. This is a package-level generic function because methods +// cannot have their own type parameters in Go (as of the current version). +// It appends the actor reference to the list associated with the key's name. +func RegisterWithReceptionist[M Message, R any](r *Receptionist, + key ServiceKey[M, R], ref ActorRef[M, R]) { + + r.mu.Lock() + defer r.mu.Unlock() + + // Initialize the slice for this key if it's the first registration. + if _, exists := r.registrations[key.name]; !exists { + r.registrations[key.name] = make([]any, 0) + } + + r.registrations[key.name] = append(r.registrations[key.name], ref) +} + +// FindInReceptionist returns all actors registered with a service key in the +// given receptionist. This is a package-level generic function because methods +// cannot have their own type parameters. It performs a type assertion to ensure +// that only ActorRefs matching the ServiceKey's generic types (M, R) are +// returned, providing type safety. +func FindInReceptionist[M Message, R any](r *Receptionist, + key ServiceKey[M, R]) []ActorRef[M, R] { + + r.mu.RLock() + defer r.mu.RUnlock() + + if refs, exists := r.registrations[key.name]; exists { + typedRefs := make([]ActorRef[M, R], 0, len(refs)) + for _, ref := range refs { + // Make sure that the reference is of the correct type. + // This type assertion is crucial for type safety, ensuring + // that the returned ActorRefs match the expected M and R. + if typedRef, ok := ref.(ActorRef[M, R]); ok { + typedRefs = append(typedRefs, typedRef) + } + } + + return typedRefs + } + + return nil +} diff --git a/actor/system_test.go b/actor/system_test.go new file mode 100644 index 000000000..cc1510ac6 --- /dev/null +++ b/actor/system_test.go @@ -0,0 +1,958 @@ +package actor + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/lightningnetwork/lnd/fn/v2" + "github.com/stretchr/testify/require" +) + +// TestActorSystemNewActorSystem verifies the basic initialization of an +// ActorSystem, including its default DLO. +func TestActorSystemNewActorSystem(t *testing.T) { + t.Parallel() + + as := NewActorSystem() + require.NotNil(t, as, "newActorSystem should not return nil") + require.NotNil(t, as.Receptionist(), "receptionist should not be nil") + require.NotNil(t, as.DeadLetters(), "deadLetters should not be nil") + require.Equal(t, "dead-letters", as.DeadLetters().ID(), "dLO ID mismatch") + + // Test the DLO's behavior (it should return an error for Ask). + testDLOMsg := newTestMsg("to-dlo") + future := as.DeadLetters().Ask(context.Background(), testDLOMsg) + result := future.Await(context.Background()) + + // We should get back an error for asks. + require.True( + t, result.IsErr(), "system DLO should return an error on Ask", + ) + expectedErrStr := "message undeliverable: " + testDLOMsg.MessageType() + require.EqualError( + t, result.Err(), expectedErrStr, "dLO error message mismatch", + ) + + // Shutdown the system to clean up resources. + err := as.Shutdown() + require.NoError(t, err, "actorSystem shutdown failed") +} + +// TestActorSystemRegisterWithSystem verifies actor registration, lifecycle +// management within the system. +func TestActorSystemRegisterWithSystem(t *testing.T) { + t.Parallel() + + as := NewActorSystem() + defer func() { + err := as.Shutdown() + require.NoError(t, err) + }() + + actorID := "test-actor-sys-reg" + serviceKey := NewServiceKey[*testMsg, string]("test-service") + + // Using echoBehavior from actor_test.go (implicitly available) + beh := newEchoBehavior(t, 0) + + // We'll start off by registering the actor. + actorRef, err := RegisterWithSystem(as, actorID, serviceKey, beh) + require.NoError(t, err) + require.NotNil(t, actorRef, "registerWithSystem should return a valid ActorRef") + require.Equal(t, actorID, actorRef.ID(), "registered actor ID mismatch") + + // The actor should be found in the receptionist. + foundActors := FindInReceptionist(as.Receptionist(), serviceKey) + require.Len(t, foundActors, 1, "actor not found in receptionist") + require.Equal(t, actorRef, foundActors[0], "incorrect actor in receptionist") + + // Next, we'll send out a simple tell, using our reply channel to make + // sure it's actually processed. + msgData := "hello-system-actor" + replyChan := make(chan string, 1) + actorRef.Tell(context.Background(), newTestMsgWithReply(msgData, replyChan)) + + received, err := fn.RecvOrTimeout(replyChan, 100*time.Millisecond) + require.NoError(t, err, "timed out waiting for actor to process message") + require.Equal(t, msgData, received, "actor did not process message") + + // Stop the actor through the system. + stopped := as.StopAndRemoveActor(actorID) + require.True(t, stopped, "StopAndRemoveActor failed") + + // Wait for actor to fully stop. + time.Sleep(50 * time.Millisecond) + + // Send a message to the now-stopped actor's ref. This should go to the + // system's DLO. + afterStopMsg := newTestMsg("after-stop-to-dlo") + require.NotPanics(t, func() { + actorRef.Tell(context.Background(), afterStopMsg) + }, "tell to stopped actor should not panic") +} + +// TestActorSystemShutdown verifies that all actors are stopped and the system +// context is cancelled upon shutdown. +func TestActorSystemShutdown(t *testing.T) { + t.Parallel() + + as := NewActorSystem() + + // We'll start by making 3 new actors, each with a unique ID. + numActors := 3 + actorRefs := make([]ActorRef[*testMsg, string], numActors) + for i := 0; i < numActors; i++ { + actorID := fmt.Sprintf("shutdown-test-actor-%d", i) + key := NewServiceKey[*testMsg, string]( + fmt.Sprintf("service-%d", i), + ) + beh := newEchoBehavior(t, 0) + ref, regErr := RegisterWithSystem(as, actorID, key, beh) + require.NoError(t, regErr) + actorRefs[i] = ref + } + + // We'll now send a message to each actor to ensure that they're + // running. + for i, ref := range actorRefs { + future := ref.Ask( + context.Background(), + newTestMsg(fmt.Sprintf("ping-%d", i)), + ) + ctxAwait, cancelAwait := context.WithTimeout( + context.Background(), time.Second, + ) + res := future.Await(ctxAwait) + cancelAwait() + require.False( + t, res.IsErr(), + "actor %d failed to respond before shutdown: %v", + i, res.Err(), + ) + } + + // Next, trigger a shutdown, and assert that the done channel gets + // closed. + err := as.Shutdown() + require.NoError(t, err, "actorSystem shutdown failed") + + // Check if the system context is done using RecvOrTimeout with a zero + // timeout for a non-blocking check. + _, err = fn.RecvOrTimeout(as.ctx.Done(), time.Millisecond*100) + require.NoError(t, err, "actorSystem context not cancelled after shutdown") + + // We'll now try to send a message to each of the actors, this should + // result in an error. + for i, ref := range actorRefs { + future := ref.Ask( + context.Background(), + newTestMsg(fmt.Sprintf("ping-after-shutdown-%d", i)), + ) + res := future.Await(context.Background()) + require.True( + t, res.IsErr(), + "actor %d Ask should fail after shutdown", i, + ) + require.ErrorIs(t, res.Err(), ErrActorTerminated) + } + + as.mu.RLock() + require.Nil(t, as.actors, "actors map should be nil after shutdown") + as.mu.RUnlock() + + // Once shutdown, we shouldn't be able to send to the DLO either. + dloRef := as.DeadLetters() + futureDLO := dloRef.Ask( + context.Background(), newTestMsg("ping-dlo-after-shutdown"), + ) + resDLO := futureDLO.Await(context.Background()) + require.True( + t, resDLO.IsErr(), "DLO Ask should fail after system shutdown", + ) + require.ErrorIs( + t, resDLO.Err(), ErrActorTerminated, + ) +} + +// TestActorSystemStopAndRemoveActor verifies specific actor stopping and +// removal. +func TestActorSystemStopAndRemoveActor(t *testing.T) { + t.Parallel() + + as := NewActorSystem() + defer func() { + err := as.Shutdown() + require.NoError(t, err) + }() + + // Make some actor IDs, then unique service keys, then use that to + // register two actors. + actor1ID := "actor-to-stop" + actor2ID := "actor-to-keep" + key1 := NewServiceKey[*testMsg, string]("service1") + key2 := NewServiceKey[*testMsg, string]("service2") + beh := newEchoBehavior(t, 0) + + ref1, err := RegisterWithSystem(as, actor1ID, key1, beh) + require.NoError(t, err) + ref2, err := RegisterWithSystem(as, actor2ID, key2, beh) + require.NoError(t, err) + + // If we remove one actor, then try to send to it, we should get an + // error. + stopped := as.StopAndRemoveActor(actor1ID) + require.True(t, stopped, "failed to stop and remove actor1") + + future1 := ref1.Ask(context.Background(), newTestMsg("ping-actor1")) + res1 := future1.Await(context.Background()) + require.True(t, res1.IsErr(), "actor1 should be stopped") + require.ErrorIs(t, res1.Err(), ErrActorTerminated) + + as.mu.RLock() + _, exists := as.actors[actor1ID] + as.mu.RUnlock() + + // The actor should no longer be found. + require.False(t, exists, "actor1 still in system's actor map") + + // Make sure that we can still send messages to the existing actor. + future2 := ref2.Ask( + context.Background(), newTestMsg("ping-actor2"), + ) + + ctxAwait2, cancelAwait2 := context.WithTimeout( + context.Background(), time.Second, + ) + res2 := future2.Await(ctxAwait2) + cancelAwait2() + + require.False( + t, res2.IsErr(), "actor2 should still be running: %v", + res2.Err(), + ) + res2.WhenOk(func(s string) { + require.Equal(t, "echo: ping-actor2", s) + }) + + stoppedNonExistent := as.StopAndRemoveActor("non-existent-actor") + require.False( + t, stoppedNonExistent, "stopping non-existent actor should "+ + "return false", + ) +} + +// TestReceptionist covers basic registration, finding, and unregistration. +func TestReceptionist(t *testing.T) { + t.Parallel() + + as := NewActorSystem() + defer func() { + err := as.Shutdown() + require.NoError(t, err) + }() + receptionist := as.Receptionist() + + key1 := NewServiceKey[*testMsg, string]("key1") + key2 := NewServiceKey[*testMsg, string]("key2") + key1Again := NewServiceKey[*testMsg, string]("key1") + + // Register 3 actor instance using the service keys we created above. + beh := newEchoBehavior(t, 0) + actor1Ref, err := RegisterWithSystem(as, "actor1-rec", key1, beh) + require.NoError(t, err) + actor2Ref, err := RegisterWithSystem(as, "actor2-rec", key1, beh) + require.NoError(t, err) + actor3Ref, err := RegisterWithSystem(as, "actor3-rec", key2, beh) + require.NoError(t, err) + + // We should be able to find the actors we registered. + foundForKey1 := FindInReceptionist(receptionist, key1) + require.Len(t, foundForKey1, 2, "should find 2 actors for key1") + require.Contains(t, foundForKey1, actor1Ref) + require.Contains(t, foundForKey1, actor2Ref) + + foundForKey1Again := FindInReceptionist(receptionist, key1Again) + require.ElementsMatch(t, foundForKey1, foundForKey1Again) + + // Same goes for the second key we added. + foundForKey2 := FindInReceptionist(receptionist, key2) + require.Len(t, foundForKey2, 1, "should find 1 actor for key2") + require.Equal(t, actor3Ref, foundForKey2[0]) + + // We shouldn't be able to find a key we didn't add. + nonExistentKey := NewServiceKey[*testMsg, string]("non-existent") + foundForNonExistent := FindInReceptionist(receptionist, nonExistentKey) + require.Empty(t, foundForNonExistent) + + // We should be able to unregister the actors we added. + unregistered := UnregisterFromReceptionist( + receptionist, key1, actor1Ref, + ) + require.True(t, unregistered, "failed to unregister actor1Ref") + + foundForKey1AfterUnreg := FindInReceptionist(receptionist, key1) + require.Len(t, foundForKey1AfterUnreg, 1) + require.Equal(t, actor2Ref, foundForKey1AfterUnreg[0]) + + // If we try to unregister the same actor again, it should fail. + unregisteredAgain := UnregisterFromReceptionist(receptionist, key1, actor1Ref) + require.False(t, unregisteredAgain) + + unregisteredLast := UnregisterFromReceptionist(receptionist, key1, actor2Ref) + require.True(t, unregisteredLast) + foundForKey1AfterAllUnreg := FindInReceptionist(receptionist, key1) + require.Empty(t, foundForKey1AfterAllUnreg) + + receptionist.mu.RLock() + _, exists := receptionist.registrations[key1.name] + receptionist.mu.RUnlock() + require.False(t, exists, "key1 should be removed from registrations map") + + // Finally, if we use the wrong key, or one that doesn't exist, that + // should also fail. + unregisteredWrongKey := UnregisterFromReceptionist(receptionist, key1, actor3Ref) + require.False(t, unregisteredWrongKey) + unregisteredNonExistentKey := UnregisterFromReceptionist(receptionist, nonExistentKey, actor1Ref) + require.False(t, unregisteredNonExistentKey) +} + +// TestServiceKeyMethods tests Spawn and Unregister methods on ServiceKey. +func TestServiceKeyMethods(t *testing.T) { + t.Parallel() + + as := NewActorSystem() + defer func() { + err := as.Shutdown() + require.NoError(t, err) + }() + + key := NewServiceKey[*testMsg, string]("sk-service") + beh := newEchoBehavior(t, 0) + + // Attempt to spawn a new actor using the service key and desired + // behavior. + actorRef, err := key.Spawn(as, "actor-sk-spawn", beh) + require.NoError(t, err) + require.NotNil(t, actorRef) + require.Equal(t, "actor-sk-spawn", actorRef.ID()) + + // We should be able to find the actor in the receptionist. + found := FindInReceptionist(as.Receptionist(), key) + require.Len(t, found, 1) + require.Equal(t, actorRef, found[0]) + + as.mu.RLock() + _, sysExists := as.actors[actorRef.ID()] + as.mu.RUnlock() + require.True(t, sysExists) + + // Next, try to unregister the actor using the service key. + success := key.Unregister(as, actorRef) + require.True(t, success, "serviceKey.Unregister failed") + + // The actor should no longer be found in the receptionist. + foundAfter := FindInReceptionist(as.Receptionist(), key) + require.Empty(t, foundAfter) + + as.mu.RLock() + _, sysExistsAfter := as.actors[actorRef.ID()] + as.mu.RUnlock() + require.False(t, sysExistsAfter) + + // If we try to send a message to the actor after unregistering it, then + // we should get an error. + future := actorRef.Ask(context.Background(), newTestMsg("ping")) + res := future.Await(context.Background()) + require.True(t, res.IsErr() && errors.Is(res.Err(), ErrActorTerminated)) + + successAgain := key.Unregister(as, actorRef) + require.False(t, successAgain) + + otherSys := NewActorSystem() // Create a different actor system + defer func() { + err := otherSys.Shutdown() + require.NoError(t, err) + }() + + // Create a dummy actor in otherSys of the correct generic type for the + // key. This actor won't be found in 'as', so Unregister should fail. + dummyBehOther := newEchoBehavior(t, 0) + dummyKeyOther := NewServiceKey[*testMsg, string]("dummy-other") + dummyActorRefOtherSys, err := RegisterWithSystem( + otherSys, "dummy-other-actor", dummyKeyOther, dummyBehOther, + ) + require.NoError(t, err) + + successNonMember := key.Unregister(as, dummyActorRefOtherSys) + require.False(t, successNonMember) +} + +// TestServiceKeyUnregisterAll tests the UnregisterAll method on ServiceKey. +// It covers scenarios including basic unregistration of multiple actors, +// attempting to unregister with no actors present, unregistering actors for +// one key while leaving others intact, and the idempotency of the operation. +func TestServiceKeyUnregisterAll(t *testing.T) { + t.Parallel() + + // Common setup for all sub-tests. + as := NewActorSystem() + defer func() { + err := as.Shutdown() + require.NoError(t, err, "ActorSystem shutdown failed.") + }() + + // Common behavior for test actors used across sub-tests. + beh := newEchoBehavior(t, 0) + + t.Run("unregister all multiple actors", func(st *testing.T) { + key1 := NewServiceKey[*testMsg, string]("sk-ua-key1") + actor1Key1, err := key1.Spawn(as, "actor1-k1-ua", beh) + require.NoError(st, err) + actor2Key1, err := key1.Spawn(as, "actor2-k1-ua", beh) + require.NoError(st, err) + + // Verify they are registered in the receptionist. + foundActorsForKey1 := FindInReceptionist( + as.Receptionist(), key1, + ) + require.Len( + st, foundActorsForKey1, 2, + "actors for key1 not in receptionist initially.", + ) + + // Verify they are in the system's actor map. + as.mu.RLock() + _, actor1Key1Exists := as.actors[actor1Key1.ID()] + _, actor2Key1Exists := as.actors[actor2Key1.ID()] + as.mu.RUnlock() + require.True( + st, actor1Key1Exists, + "actor1 for key1 not in system actors map initially.", + ) + require.True( + st, actor2Key1Exists, + "actor2 for key1 not in system actors map initially.", + ) + + // Unregister all for key1. + stoppedCountKey1 := key1.UnregisterAll(as) + require.Equal( + st, 2, stoppedCountKey1, + "UnregisterAll for key1 returned incorrect count.", + ) + + // Verify they are unregistered from the receptionist. + foundActorsForKey1After := FindInReceptionist( + as.Receptionist(), key1, + ) + require.Empty( + st, foundActorsForKey1After, + "actors for key1 still in receptionist after "+ + "UnregisterAll.", + ) + + // Verify they are removed from system actors map. + as.mu.RLock() + _, actor1Key1ExistsAfter := as.actors[actor1Key1.ID()] + _, actor2Key1ExistsAfter := as.actors[actor2Key1.ID()] + as.mu.RUnlock() + require.False( + st, actor1Key1ExistsAfter, + "Actor1 for key1 still in system actors "+ + "map after UnregisterAll.", + ) + require.False( + st, actor2Key1ExistsAfter, + "Actor2 for key1 still in system actors "+ + "map after UnregisterAll.", + ) + + // Verify actors are stopped. + resultActor1Key1 := actor1Key1.Ask( + context.Background(), newTestMsg("ping-k1-a1"), + ).Await(context.Background()) + require.True( + st, resultActor1Key1.IsErr(), + "Actor1 key1 Ask should fail after UnregisterAll.", + ) + require.ErrorIs( + st, resultActor1Key1.Err(), ErrActorTerminated, + "Actor1 key1 not terminated with correct error.", + ) + + resultActor2Key1 := actor2Key1.Ask( + context.Background(), newTestMsg("ping-k1-a2"), + ).Await(context.Background()) + require.True( + st, resultActor2Key1.IsErr(), + "Actor2 key1 Ask should fail after UnregisterAll.", + ) + require.ErrorIs( + st, resultActor2Key1.Err(), ErrActorTerminated, + "Actor2 key1 not terminated with correct error.", + ) + }) + + t.Run("unregister all with no actors for the key", func(st *testing.T) { + keyEmpty := NewServiceKey[*testMsg, string]("sk-ua-key-empty") + stoppedCountEmptyKey := keyEmpty.UnregisterAll(as) + require.Equal( + st, 0, stoppedCountEmptyKey, + "UnregisterAll for empty key returned non-zero count.", + ) + + foundActorsForKeyEmpty := FindInReceptionist( + as.Receptionist(), keyEmpty, + ) + require.Empty( + st, foundActorsForKeyEmpty, + "Receptionist not empty for keyEmpty "+ + "after UnregisterAll.", + ) + }) + + t.Run("unregister all with mixed keys", func(st *testing.T) { + keyA := NewServiceKey[*testMsg, string]("sk-ua-keyA") + keyB := NewServiceKey[*testMsg, string]("sk-ua-keyB") + + // Spawn 3 actors, two of them will share the same service key. + actorA1, err := keyA.Spawn(as, "actorA1-ua-mixed", beh) + require.NoError(st, err) + actorA2, err := keyA.Spawn(as, "actorA2-ua-mixed", beh) + require.NoError(st, err) + actorB1, err := keyB.Spawn(as, "actorB1-ua-mixed", beh) + require.NoError(st, err) + + // Make sure we're able to find them in the receptionist. + require.Len( + st, FindInReceptionist(as.Receptionist(), keyA), 2, + "KeyA initial registration count mismatch.", + ) + require.Len( + st, FindInReceptionist(as.Receptionist(), keyB), 1, + "KeyB initial registration count mismatch.", + ) + + // We'll start by unregistering all actors for keyA. + stoppedCountKeyA := keyA.UnregisterAll(as) + require.Equal( + st, 2, stoppedCountKeyA, + "UnregisterAll for keyA returned incorrect count.", + ) + + // Verify keyA actors are gone from receptionist, keyB actor + // remains. + require.Empty( + st, FindInReceptionist(as.Receptionist(), keyA), + "actors for keyA still in receptionist after "+ + "UnregisterAll.", + ) + foundActorsForKeyBAfterA := FindInReceptionist( + as.Receptionist(), keyB, + ) + require.Len( + st, foundActorsForKeyBAfterA, 1, + "Actor for keyB affected by UnregisterAll on keyA.", + ) + require.Equal( + st, actorB1, foundActorsForKeyBAfterA[0], + "Wrong actor found for keyB.", + ) + + // Verify keyA actors are removed from system map, keyB actor + // remains. + as.mu.RLock() + _, actorA1ExistsAfterMixed := as.actors[actorA1.ID()] + _, actorA2ExistsAfterMixed := as.actors[actorA2.ID()] + _, actorB1ExistsAfterMixed := as.actors[actorB1.ID()] + as.mu.RUnlock() + require.False( + st, actorA1ExistsAfterMixed, + "ActorA1 still in system actors map after "+ + "mixed UnregisterAll.", + ) + require.False( + st, actorA2ExistsAfterMixed, + "ActorA2 still in system actors map after "+ + "mixed UnregisterAll.", + ) + require.True( + st, actorB1ExistsAfterMixed, + "ActorB1 removed from system actors map incorrectly.", + ) + + // Verify keyA actors are stopped, keyB actor is running. + resultActorA1Mixed := actorA1.Ask( + context.Background(), newTestMsg("ping-kA-a1"), + ).Await(context.Background()) + require.True(st, resultActorA1Mixed.IsErr()) + require.ErrorIs( + st, resultActorA1Mixed.Err(), ErrActorTerminated, + ) + + resultActorB1Mixed := actorB1.Ask( + context.Background(), newTestMsg("ping-kB-a1"), + ).Await(context.Background()) + require.False( + st, resultActorB1Mixed.IsErr(), + "ActorB1 terminated incorrectly (mixed test): %v", + resultActorB1Mixed.Err(), + ) + resultActorB1Mixed.WhenOk(func(s string) { + require.Equal(st, "echo: ping-kB-a1", s) + }) + }) + + t.Run("idempotency of UnregisterAll", func(st *testing.T) { + keyIdempotent := NewServiceKey[*testMsg, string]( + "sk-ua-key-idem", + ) + actorIdem, err := keyIdempotent.Spawn(as, "actor-idem-ua", beh) + require.NoError(st, err) + + // First call should unregister and stop. + stoppedCountFirstCall := keyIdempotent.UnregisterAll(as) + require.Equal( + st, 1, stoppedCountFirstCall, + "UnregisterAll (first call) incorrect count.", + ) + + // Second call should do nothing and return 0. + stoppedCountSecondCall := keyIdempotent.UnregisterAll(as) + require.Equal( + st, 0, stoppedCountSecondCall, + "UnregisterAll (second call) incorrect count, not "+ + "idempotent.", + ) + + // Verify actor is gone from receptionist and system map, and is + // stopped. + require.Empty( + st, FindInReceptionist(as.Receptionist(), keyIdempotent), + "Actors for keyIdempotent still in receptionist "+ + "after calls.", + ) + + as.mu.RLock() + _, actorIdemExistsAfter := as.actors[actorIdem.ID()] + as.mu.RUnlock() + require.False( + st, actorIdemExistsAfter, + "ActorIdem still in system actors map after calls.", + ) + + resultActorIdem := actorIdem.Ask( + context.Background(), newTestMsg("ping-kidem-a1"), + ).Await(context.Background()) + require.True(st, resultActorIdem.IsErr()) + require.ErrorIs(st, resultActorIdem.Err(), ErrActorTerminated) + }) +} + +// routerTestHarness helps set up routers and their associated actors for testing. +// It uses an actorTestHarness internally for DLO observation for the router. +type routerTestHarness struct { + *actorTestHarness + as *ActorSystem + receptionist *Receptionist +} + +// newRouterTestHarness sets up a new harness for router testing. +// It creates an ActorSystem for actors that the router will route to, +// and uses the embedded actorTestHarness for the router's own DLO. +func newRouterTestHarness(t *testing.T) *routerTestHarness { + t.Helper() + system := NewActorSystem() + t.Cleanup(func() { + err := system.Shutdown() + require.NoError(t, err, "router test actor system shutdown failed") + }) + + // The DLO for the router itself will come from actorTestHarness. + // Actors managed by `system` (router targets) will use `system.DeadLetters()`. + return &routerTestHarness{ + actorTestHarness: newActorTestHarness(t), + as: system, + receptionist: system.Receptionist(), + } +} + +// newRouterTargetActor creates an actor, registers it with the harness's +// ActorSystem (h.as) and Receptionist under the given service key. This actor +// is intended to be a target for the router. +func (h *routerTestHarness) newRouterTargetActor(id string, + key ServiceKey[*testMsg, string], + beh ActorBehavior[*testMsg, string]) ActorRef[*testMsg, string] { + + h.t.Helper() + + ref, err := RegisterWithSystem(h.as, id, key, beh) + require.NoError(h.t, err) + + return ref +} + +// TestRouterNewRouter verifies that a new router can be created as expected. +func TestRouterNewRouter(t *testing.T) { + t.Parallel() + h := newRouterTestHarness(t) + + key := NewServiceKey[*testMsg, string]("router-service") + strategy := NewRoundRobinStrategy[*testMsg, string]() + + router := NewRouter(h.receptionist, key, strategy, h.dlo.Ref()) + require.NotNil(t, router, "newRouter should not return nil") + require.Equal(t, "router(router-service)", router.ID(), "router ID mismatch") +} + +// countingEchoBehavior is an echo behavior that also counts how many messages +// it has processed. +type countingEchoBehavior struct { + *echoBehavior + id string + processedMsgs int64 +} + +func newCountingEchoBehavior(t *testing.T, id string) *countingEchoBehavior { + return &countingEchoBehavior{ + echoBehavior: newEchoBehavior(t, 0), + id: id, + } +} + +func (b *countingEchoBehavior) Receive(ctx context.Context, + msg *testMsg) fn.Result[string] { + + atomic.AddInt64(&b.processedMsgs, 1) + + // Include actor ID in reply for easier verification. + res := b.echoBehavior.Receive(ctx, msg) + val, err := res.Unpack() + if err == nil { + return fn.Ok(fmt.Sprintf("%s:%s", b.id, val)) + } + return res +} + +// TestRouterTellAndAskRoundRobin verifies that the router distributes messages +// in a round robin properly. +func TestRouterTellAndAskRoundRobin(t *testing.T) { + t.Parallel() + h := newRouterTestHarness(t) + + // Make a new router for the given service key and round robin strategy. + serviceKey := NewServiceKey[*testMsg, string]("rr-service") + strategy := NewRoundRobinStrategy[*testMsg, string]() + router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref()) + + // We'll now register two actors with the router, each with a different + // service key. + actor1Beh := newCountingEchoBehavior(t, "actor1") + actor2Beh := newCountingEchoBehavior(t, "actor2") + _ = h.newRouterTargetActor("actor1-rr", serviceKey, actor1Beh) + _ = h.newRouterTargetActor("actor2-rr", serviceKey, actor2Beh) + + // Nxet, we'll send a mix of Tell and Ask messages to the router. + numMessages := 6 + for i := 0; i < numMessages; i++ { + msgData := fmt.Sprintf("message-%d", i) + if i%2 == 0 { + router.Tell(context.Background(), newTestMsg(msgData)) + } else { + future := router.Ask( + context.Background(), newTestMsg(msgData), + ) + ctxAwait, cancelAwait := context.WithTimeout( + context.Background(), time.Second, + ) + + result := future.Await(ctxAwait) + cancelAwait() + require.False( + t, result.IsErr(), "ask failed: %v", result.Err(), + ) + } + } + + // Wait a bit for Tell messages to be processed. + time.Sleep(100 * time.Millisecond) + + // Each actor should have processed numMessages / 2 messages. + require.EqualValues( + t, numMessages/2, atomic.LoadInt64(&actor1Beh.processedMsgs), + "actor1 processed message count mismatch", + ) + require.EqualValues( + t, numMessages/2, atomic.LoadInt64(&actor2Beh.processedMsgs), + "actor2 processed message count mismatch", + ) + + // Router's DLO should be empty. + h.assertNoDLOMessages() +} + +// TestRouterNoActorsAvailable verifies that if no actors are available for the +// message, then an error is returned. +func TestRouterNoActorsAvailable(t *testing.T) { + t.Parallel() + h := newRouterTestHarness(t) + + serviceKey := NewServiceKey[*testMsg, string]("no-actor-service") + strategy := NewRoundRobinStrategy[*testMsg, string]() + router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref()) + + // We'll send a message, then assert that it goes to the DLO. + tellMsg := newTestMsg("tell-no-actor") + router.Tell(context.Background(), tellMsg) + h.assertDLOMessage(tellMsg) + + // If we use an ask instead, then we should get an error. + askMsg := newTestMsg("ask-no-actor") + future := router.Ask(context.Background(), askMsg) + result := future.Await(context.Background()) + + require.True( + t, result.IsErr(), "ask should fail when no actors are available", + ) + require.ErrorIs(t, result.Err(), ErrNoActorsAvailable, "error mismatch") +} + +// TestRouterTellAskContextCancellation verifies that if the context is +// canceled, then sending aborts. +func TestRouterTellAskContextCancellation(t *testing.T) { + t.Parallel() + h := newRouterTestHarness(t) + + serviceKey := NewServiceKey[*testMsg, string]("ctx-cancel-service") + strategy := NewRoundRobinStrategy[*testMsg, string]() + router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref()) + + // Use a regular echo actor, but we'll control context for Tell/Ask. + targetActorBeh := newEchoBehavior(t, 50*time.Millisecond) + _ = h.newRouterTargetActor("target-ctx", serviceKey, targetActorBeh) + + // Next, we'll send a Tell message with a context that will be cancelled + // before we even send. + ctxTell, cancelTell := context.WithCancel(context.Background()) + cancelTell() + router.Tell(ctxTell, newTestMsg("tell-ctx-cancelled")) + + // The Message should be dropped by actorRefImpl.Tell if ctx is + // cancelled. Router's DLO should not receive it from this path. + h.assertNoDLOMessages() + + // Next, we'll do the same for Ask. This time, we should get an error. + ctxAsk, cancelAsk := context.WithCancel(context.Background()) + cancelAsk() + futureAsk := router.Ask(ctxAsk, newTestMsg("ask-ctx-cancelled")) + resultAsk := futureAsk.Await(context.Background()) + + require.True( + t, resultAsk.IsErr(), "ask with cancelled context should fail", + ) + require.ErrorIs( + t, resultAsk.Err(), context.Canceled, + "error should be context.Canceled", + ) +} + +// TestRouterDynamicActorRegistration tests that we're able to dynamically add +// and remove actors from the router. +func TestRouterDynamicActorRegistration(t *testing.T) { + t.Parallel() + h := newRouterTestHarness(t) + + serviceKey := NewServiceKey[*testMsg, string]("dynamic-service") + strategy := NewRoundRobinStrategy[*testMsg, string]() + router := NewRouter(h.receptionist, serviceKey, strategy, h.dlo.Ref()) + + // If we try to send a mesasge to the router before any actors are + // added, we should get an error. + futureNoActor := router.Ask(context.Background(), newTestMsg("ping-no-actors")) + resNoActor := futureNoActor.Await(context.Background()) + require.ErrorIs(t, resNoActor.Err(), ErrNoActorsAvailable) + + actor1Beh := newCountingEchoBehavior(t, "actor1") + actor1Ref := h.newRouterTargetActor("actor1-dynamic", serviceKey, actor1Beh) + + // At this point, we have a new actor added, but we'll try to send a + // message to a different actor ID. This should go to the router's DLO. + futureActor1 := router.Ask(context.Background(), newTestMsg("ping-actor1")) + ctxAwaitA1, cancelAwaitA1 := context.WithTimeout(context.Background(), time.Second) + resActor1 := futureActor1.Await(ctxAwaitA1) + cancelAwaitA1() + require.False(t, resActor1.IsErr(), "ask to actor1 failed: %v", resActor1.Err()) + resActor1.WhenOk(func(s string) { + require.Equal(t, "actor1:echo: ping-actor1", s) + }) + + actor2Beh := newCountingEchoBehavior(t, "actor2") + actor2Ref := h.newRouterTargetActor( + "actor2-dynamic", serviceKey, actor2Beh, + ) + + // Now that we've added two actors above, we should round robin between + // them when sending. + ctxAwaitDA1, cancelAwaitDA1 := context.WithTimeout( + context.Background(), time.Second, + ) + router.Ask(context.Background(), newTestMsg("dynamic-ask1")).Await( + ctxAwaitDA1, + ) + cancelAwaitDA1() + + ctxAwaitDA2, cancelAwaitDA2 := context.WithTimeout(context.Background(), time.Second) + router.Ask(context.Background(), newTestMsg("dynamic-ask2")).Await(ctxAwaitDA2) + cancelAwaitDA2() + + time.Sleep(50 * time.Millisecond) + + // actor1 should have processed 2 messages (ping-actor1, dynamic-ask1), + require.EqualValues(t, 2, atomic.LoadInt64(&actor1Beh.processedMsgs)) + require.EqualValues(t, 1, atomic.LoadInt64(&actor2Beh.processedMsgs)) + + // Next, we'll unregister the first actor ref. + unregistered := UnregisterFromReceptionist( + h.receptionist, serviceKey, actor1Ref, + ) + require.True(t, unregistered) + + // All the messages should now go to the second actor. + for i := 0; i < 2; i++ { + msgData := fmt.Sprintf("to-actor2-%d", i) + future := router.Ask(context.Background(), newTestMsg(msgData)) + ctxAwaitLoop, cancelAwaitLoop := context.WithTimeout( + context.Background(), time.Second, + ) + + res := future.Await(ctxAwaitLoop) + cancelAwaitLoop() + + require.False( + t, res.IsErr(), "ask to actor2 failed: %v", res.Err(), + ) + res.WhenOk(func(s string) { + require.Equal(t, "actor2:echo: "+msgData, s) + }) + } + + // Actor 1 shouldn't have got any of the messages, they should go to + // actor 2. + require.EqualValues(t, 2, atomic.LoadInt64(&actor1Beh.processedMsgs)) + require.EqualValues(t, 1+2, atomic.LoadInt64(&actor2Beh.processedMsgs)) + + // Next, we'll unregister the second actor ref. + unregistered2 := UnregisterFromReceptionist( + h.receptionist, serviceKey, actor2Ref, + ) + require.True(t, unregistered2) + + // If we try to send another message, it should go to the DL. + tellMsg := newTestMsg("dynamic-tell-no-actors") + router.Tell(context.Background(), tellMsg) + h.assertDLOMessage(tellMsg) +} From 7c9d72134c441ef8e28cfc1a28f00876fd840b7e Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:58:25 -0700 Subject: [PATCH 5/6] actor: add example files In this commit, we add a series of examples that show how the package can be used in the wild. They can be run as normal Example tests. --- actor/example_basic_actor_test.go | 101 +++++++++++++++++++ actor/example_router_test.go | 120 ++++++++++++++++++++++ actor/example_struct_actor_test.go | 153 +++++++++++++++++++++++++++++ actor/example_tell_only_test.go | 137 ++++++++++++++++++++++++++ 4 files changed, 511 insertions(+) create mode 100644 actor/example_basic_actor_test.go create mode 100644 actor/example_router_test.go create mode 100644 actor/example_struct_actor_test.go create mode 100644 actor/example_tell_only_test.go diff --git a/actor/example_basic_actor_test.go b/actor/example_basic_actor_test.go new file mode 100644 index 000000000..960602a91 --- /dev/null +++ b/actor/example_basic_actor_test.go @@ -0,0 +1,101 @@ +package actor_test + +import ( + "context" + "fmt" + "time" + + "github.com/lightningnetwork/lnd/actor" + "github.com/lightningnetwork/lnd/fn/v2" +) + +// BasicGreetingMsg is a simple message type for the basic actor example. +type BasicGreetingMsg struct { + actor.BaseMessage + Name string +} + +// MessageType implements actor.Message. +func (m BasicGreetingMsg) MessageType() string { return "BasicGreetingMsg" } + +// BasicGreetingResponse is a simple response type. +type BasicGreetingResponse struct { + Greeting string +} + +// ExampleActor demonstrates creating a single actor, sending it a message +// directly using Ask, and then unregistering and stopping it. +func ExampleActor() { + system := actor.NewActorSystem() + defer system.Shutdown() + + //nolint:ll + greeterKey := actor.NewServiceKey[BasicGreetingMsg, BasicGreetingResponse]( + "basic-greeter", + ) + + actorID := "my-greeter" + greeterBehavior := actor.NewFunctionBehavior( + func(ctx context.Context, + msg BasicGreetingMsg) fn.Result[BasicGreetingResponse] { + + return fn.Ok(BasicGreetingResponse{ + Greeting: "Hello, " + msg.Name + " from " + + actorID, + }) + }, + ) + + // Spawn the actor. This registers it with the system and receptionist, + // and starts it. It returns an ActorRef. + greeterRef, err := greeterKey.Spawn(system, actorID, greeterBehavior) + if err != nil { + fmt.Printf("Failed to spawn actor: %v\n", err) + return + } + fmt.Printf("Actor %s spawned.\n", greeterRef.ID()) + + // Send a message directly to the actor's reference. + askCtx, askCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + defer askCancel() + futureResponse := greeterRef.Ask( + askCtx, BasicGreetingMsg{Name: "World"}, + ) + + awaitCtx, awaitCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + defer awaitCancel() + result := futureResponse.Await(awaitCtx) + + result.WhenErr(func(err error) { + fmt.Printf("Error awaiting response: %v\n", err) + }) + result.WhenOk(func(response BasicGreetingResponse) { + fmt.Printf("Received: %s\n", response.Greeting) + }) + + // Unregister the actor. This also stops the actor. + unregistered := greeterKey.Unregister(system, greeterRef) + if unregistered { + fmt.Printf("Actor %s unregistered and stopped.\n", + greeterRef.ID()) + } else { + fmt.Printf("Failed to unregister actor %s.\n", greeterRef.ID()) + } + + // Verify it's no longer in the receptionist. + refsAfterUnregister := actor.FindInReceptionist( + system.Receptionist(), greeterKey, + ) + fmt.Printf("Actors for key '%s' after unregister: %d\n", + "basic-greeter", len(refsAfterUnregister)) + + // Output: + // Actor my-greeter spawned. + // Received: Hello, World from my-greeter + // Actor my-greeter unregistered and stopped. + // Actors for key 'basic-greeter' after unregister: 0 +} diff --git a/actor/example_router_test.go b/actor/example_router_test.go new file mode 100644 index 000000000..f68cff60a --- /dev/null +++ b/actor/example_router_test.go @@ -0,0 +1,120 @@ +package actor_test + +import ( + "context" + "fmt" + "time" + + "github.com/lightningnetwork/lnd/actor" + "github.com/lightningnetwork/lnd/fn/v2" +) + +// RouterGreetingMsg is a message type for the router example. +type RouterGreetingMsg struct { + actor.BaseMessage + Name string +} + +// MessageType implements actor.Message. +func (m RouterGreetingMsg) MessageType() string { return "RouterGreetingMsg" } + +// RouterGreetingResponse is a response type for the router example. +type RouterGreetingResponse struct { + Greeting string + HandlerID string +} + +// ExampleRouter demonstrates creating multiple actors under the same service +// key and using a router to dispatch messages to them. +func ExampleRouter() { + system := actor.NewActorSystem() + defer system.Shutdown() + + //nolint:ll + routerGreeterKey := actor.NewServiceKey[RouterGreetingMsg, RouterGreetingResponse]( + "router-greeter-service", + ) + + // Behavior for the first greeter actor. + actorID1 := "router-greeter-1" + greeterBehavior1 := actor.NewFunctionBehavior( + func(ctx context.Context, + msg RouterGreetingMsg) fn.Result[RouterGreetingResponse] { + + return fn.Ok(RouterGreetingResponse{ + Greeting: "Greetings, " + msg.Name + "!", + HandlerID: actorID1, + }) + }, + ) + _, err := routerGreeterKey.Spawn(system, actorID1, greeterBehavior1) + if err != nil { + fmt.Printf("Failed to spawn actor: %v\n", err) + return + } + fmt.Printf("Actor %s spawned.\n", actorID1) + + // Behavior for the second greeter actor. + actorID2 := "router-greeter-2" + greeterBehavior2 := actor.NewFunctionBehavior( + func(ctx context.Context, + msg RouterGreetingMsg) fn.Result[RouterGreetingResponse] { + + return fn.Ok(RouterGreetingResponse{ + Greeting: "Salutations, " + msg.Name + "!", + HandlerID: actorID2, + }) + }, + ) + _, err = routerGreeterKey.Spawn(system, actorID2, greeterBehavior2) + if err != nil { + fmt.Printf("Failed to spawn actor: %v\n", err) + return + } + fmt.Printf("Actor %s spawned.\n", actorID2) + + // Create a router for the "router-greeter-service". + greeterRouter := actor.NewRouter( + system.Receptionist(), routerGreeterKey, + actor.NewRoundRobinStrategy[RouterGreetingMsg, + RouterGreetingResponse](), + system.DeadLetters(), + ) + fmt.Printf("Router %s created for service key '%s'.\n", + greeterRouter.ID(), "router-greeter-service") + + // Send messages through the router. + names := []string{"Alice", "Bob", "Charlie", "David"} + for _, name := range names { + askCtx, askCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + futureResponse := greeterRouter.Ask( + askCtx, RouterGreetingMsg{Name: name}, + ) + + awaitCtx, awaitCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + result := futureResponse.Await(awaitCtx) + + result.WhenErr(func(err error) { + fmt.Printf("For %s: Error - %v\n", name, err) + }) + result.WhenOk(func(response RouterGreetingResponse) { + fmt.Printf("For %s: Received '%s' from %s\n", + name, response.Greeting, response.HandlerID) + }) + awaitCancel() + askCancel() + } + + // Output: + // Actor router-greeter-1 spawned. + // Actor router-greeter-2 spawned. + // Router router(router-greeter-service) created for service key 'router-greeter-service'. + // For Alice: Received 'Greetings, Alice!' from router-greeter-1 + // For Bob: Received 'Salutations, Bob!' from router-greeter-2 + // For Charlie: Received 'Greetings, Charlie!' from router-greeter-1 + // For David: Received 'Salutations, David!' from router-greeter-2 +} diff --git a/actor/example_struct_actor_test.go b/actor/example_struct_actor_test.go new file mode 100644 index 000000000..e92e1e1a7 --- /dev/null +++ b/actor/example_struct_actor_test.go @@ -0,0 +1,153 @@ +package actor_test + +import ( + "context" + "fmt" + "time" + + "github.com/lightningnetwork/lnd/actor" + "github.com/lightningnetwork/lnd/fn/v2" +) + +// CounterMsg is a message type for the stateful counter actor. +// It can be used to increment the counter or get its current value. +type CounterMsg struct { + actor.BaseMessage + Increment int + GetValue bool + Who string +} + +// MessageType implements actor.Message. +func (m CounterMsg) MessageType() string { return "CounterMsg" } + +// CounterResponse is a response type for the counter actor. +type CounterResponse struct { + Value int + Responder string +} + +// StatefulCounterActor demonstrates an actor that maintains internal state (a +// counter) and processes messages to modify or query that state. +type StatefulCounterActor struct { + counter int + actorID string +} + +// NewStatefulCounterActor creates a new counter actor. +func NewStatefulCounterActor(id string) *StatefulCounterActor { + return &StatefulCounterActor{ + actorID: id, + } +} + +// Receive is the message handler for the StatefulCounterActor. +// It implements the actor.ActorBehavior interface implicitly when wrapped. +func (s *StatefulCounterActor) Receive(ctx context.Context, + msg CounterMsg) fn.Result[CounterResponse] { + + if msg.Increment > 0 { + // For increment, we can just acknowledge or return the new + // value. Messages are sent serially, so we don't need to worry + // about a mutex here. + s.counter += msg.Increment + + return fn.Ok(CounterResponse{ + Value: s.counter, + Responder: s.actorID, + }) + } + + if msg.GetValue { + return fn.Ok(CounterResponse{ + Value: s.counter, + Responder: s.actorID, + }) + } + + return fn.Err[CounterResponse](fmt.Errorf("invalid CounterMsg")) +} + +// ExampleActor_stateful demonstrates creating an actor whose behavior is defined +// by a struct with methods, allowing it to maintain internal state. +func ExampleActor_stateful() { + system := actor.NewActorSystem() + defer system.Shutdown() + + counterServiceKey := actor.NewServiceKey[CounterMsg, CounterResponse]( + "struct-counter-service", + ) + + // Create an instance of our stateful actor logic. + actorID := "counter-actor-1" + counterLogic := NewStatefulCounterActor(actorID) + + // Spawn the actor. + // The counterLogic instance itself satisfies the ActorBehavior + // interface because its Receive method matches the required signature. + counterRef, err := counterServiceKey.Spawn( + system, actorID, counterLogic, + ) + if err != nil { + fmt.Printf("Failed to spawn actor: %v\n", err) + return + } + fmt.Printf("Actor %s spawned.\n", counterRef.ID()) + + // Send messages to increment the counter. + for i := 1; i <= 3; i++ { + askCtx, askCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + futureResp := counterRef.Ask(askCtx, + CounterMsg{ + Increment: i, + Who: fmt.Sprintf("Incrementer-%d", i), + }, + ) + awaitCtx, awaitCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + resp := futureResp.Await(awaitCtx) + + resp.WhenOk(func(r CounterResponse) { + fmt.Printf("Incremented by %d, new value: %d "+ + "(from %s)\n", i, r.Value, r.Responder) + }) + resp.WhenErr(func(e error) { + fmt.Printf("Error incrementing: %v\n", e) + }) + awaitCancel() + askCancel() + } + + // Send a message to get the current value. + askCtx, askCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + futureResp := counterRef.Ask( + askCtx, CounterMsg{GetValue: true, Who: "Getter"}, + ) + + awaitCtx, awaitCancel := context.WithTimeout( + context.Background(), 1*time.Second, + ) + + finalValueResp := futureResp.Await(awaitCtx) + finalValueResp.WhenOk(func(r CounterResponse) { + fmt.Printf("Final counter value: %d (from %s)\n", + r.Value, r.Responder) + }) + finalValueResp.WhenErr(func(e error) { + fmt.Printf("Error getting value: %v\n", e) + }) + awaitCancel() + askCancel() + + // Output: + // Actor counter-actor-1 spawned. + // Incremented by 1, new value: 1 (from counter-actor-1) + // Incremented by 2, new value: 3 (from counter-actor-1) + // Incremented by 3, new value: 6 (from counter-actor-1) + // Final counter value: 6 (from counter-actor-1) +} diff --git a/actor/example_tell_only_test.go b/actor/example_tell_only_test.go new file mode 100644 index 000000000..a6371310b --- /dev/null +++ b/actor/example_tell_only_test.go @@ -0,0 +1,137 @@ +package actor_test + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + "github.com/lightningnetwork/lnd/actor" + "github.com/lightningnetwork/lnd/fn/v2" +) + +// LogMsg is a message type for the TellOnly example. +type LogMsg struct { + actor.BaseMessage + Text string +} + +// MessageType implements actor.Message. +func (m LogMsg) MessageType() string { return "LogMsg" } + +// LoggerActorBehavior is a simple actor behavior that logs messages. It doesn't +// produce a meaningful response for Ask, so it's a good candidate for TellOnly +// interactions. +type LoggerActorBehavior struct { + mu sync.Mutex + logs []string + actorID string +} + +func NewLoggerActorBehavior(id string) *LoggerActorBehavior { + return &LoggerActorBehavior{actorID: id} +} + +// Receive processes LogMsg messages by appending them to an internal log. The +// response type is 'any' as it's not typically used with Ask. +func (l *LoggerActorBehavior) Receive(ctx context.Context, + msg actor.Message) fn.Result[any] { + + logMessage, ok := msg.(LogMsg) + if !ok { + return fn.Err[any](fmt.Errorf("unexpected message "+ + "type: %s", msg.MessageType())) + } + + l.mu.Lock() + defer l.mu.Unlock() + + entry := fmt.Sprintf("[%s from %s]: %s", time.Now().Format("15:04:05"), + l.actorID, logMessage.Text) + l.logs = append(l.logs, entry) + + // For Tell, the result is often ignored, but we must return something. + return fn.Ok[any](nil) +} + +func (l *LoggerActorBehavior) GetLogs() []string { + l.mu.Lock() + defer l.mu.Unlock() + + copiedLogs := make([]string, len(l.logs)) + copy(copiedLogs, l.logs) + + return copiedLogs +} + +// ExampleTellOnlyRef demonstrates using a TellOnlyRef for fire-and-forget +// messaging with an actor. +func ExampleTellOnlyRef() { + system := actor.NewActorSystem() + defer system.Shutdown() + + // The logger actor doesn't really have a response type for Ask, so we + // use 'any'. + loggerServiceKey := actor.NewServiceKey[actor.Message, any]( + "tell-only-logger-service", + ) + + actorID := "my-logger" + loggerLogic := NewLoggerActorBehavior(actorID) + + // Spawn the actor. + fullRef, err := loggerServiceKey.Spawn(system, actorID, loggerLogic) + if err != nil { + fmt.Printf("Failed to spawn actor: %v\n", err) + return + } + fmt.Printf("Actor %s spawned.\n", fullRef.ID()) + + // Get a TellOnlyRef for the actor. We can get this from the Actor + // instance itself if we had it, or by type assertion if we know the + // underlying ref supports it. Since fullRef is ActorRef[actor.Message, + // any], it already satisfies TellOnlyRef[actor.Message]. + // + // Or, if we had the *Actor instance: tellOnlyLogger = + // actorInstance.TellRef() + var tellOnlyLogger actor.TellOnlyRef[actor.Message] = fullRef + + fmt.Printf("Obtained TellOnlyRef for %s.\n", tellOnlyLogger.ID()) + + // Send messages using Tell. + tellOnlyLogger.Tell( + context.Background(), LogMsg{Text: "First log entry."}, + ) + tellOnlyLogger.Tell( + context.Background(), LogMsg{Text: "Second log entry."}, + ) + + // Allow some time for messages to be processed. + time.Sleep(10 * time.Millisecond) + + // Retrieve logs directly from the behavior for verification in this + // example. In a real scenario, this might not be possible or desired. + logs := loggerLogic.GetLogs() + fmt.Println("Logged entries:") + for _, entry := range logs { + // Strip the timestamp and actor ID for consistent example + // output. Example entry: "[15:04:05 from my-logger]: Actual log + // text" + parts := strings.SplitN(entry, "]: ", 2) + if len(parts) == 2 { + fmt.Println(parts[1]) + } + } + + // Attempting to Ask using tellOnlyLogger would be a compile-time error: + // tellOnlyLogger.Ask(context.Background(), LogMsg{Text: "This would + // fail"}) + + // Output: + // Actor my-logger spawned. + // Obtained TellOnlyRef for my-logger. + // Logged entries: + // First log entry. + // Second log entry. +} From 8781daa995bc007cb8ba1f96c0fef672cea4d5ee Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:58:58 -0700 Subject: [PATCH 6/6] actor: add README.md In this commit, we add a readme which serves as a general introduction to the pacakge, and also the motivation of the package. It serves as a manual for developers that may wish to interact with the package. --- actor/README.md | 478 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 478 insertions(+) create mode 100644 actor/README.md diff --git a/actor/README.md b/actor/README.md new file mode 100644 index 000000000..0d138bb69 --- /dev/null +++ b/actor/README.md @@ -0,0 +1,478 @@ +# Actor Package + +## Introduction to Actors + +The actor model is a conceptual model for concurrent computation that treats +"actors" as the universal primitives of concurrent computation. Originating from +Carl Hewitt's work in the 1970s and popularized by languages like Erlang and +frameworks like Akka, actors provide a high-level abstraction for building +robust, concurrent, and distributed systems. + +At its core, an actor is an independent unit of computation that encapsulates: +- **State**: An actor can maintain private state that it alone can modify. +- **Behavior**: An actor defines how it reacts to messages it receives. +- **Mailbox**: Each actor has a mailbox to queue incoming messages. + +Actors communicate exclusively through asynchronous message passing. When an +actor receives a message, it can: +1. Send a finite number of messages to other actors. +2. Create a finite number of new actors. +3. Designate the behavior to be used for the next message it receives (which + can be the same behavior). + +Concurrency is managed by the actor system, allowing many actors to execute +concurrently without explicit lock management by the developer for actor state. +This model inherently promotes loose coupling, as actors do not share state and +interact only through messages. + +## Motivation for this Package + +In large, long-lived systems like `lnd`, managing complexity, concurrency, and +component lifecycles becomes increasingly challenging. This `actor` package is +introduced to address several key motivations: + +### Structured Message Passing + +To move away from direct, synchronous method calls between major components, +especially where concurrency or complex state interactions are involved. Message +passing encourages clearer, more auditable interactions and helps manage +concurrent access to component state. + +### Eliminating "God Structs" + +Over time, systems can develop large "god structs" that hold references to +numerous sub-systems. This leads to tight coupling, makes dependency management +difficult, and can obscure the flow of control and data. Actors, by +encapsulating state and behavior and interacting via messages, help break down +these monolithic structures into more manageable, independent units. + +### Decoupled Lifecycles + +Often, the lifecycle of a sub-system is unnecessarily tied to a parent system, +or access to a sub-system requires traversing through a central "manager" +object. Actors can have independent lifecycles managed by an actor system, +allowing for more granular control over starting, stopping, and restarting +components. + +An example of such interaction is when an RPC call needs to go through several +other structs to obtain a reference to a given sub-system, in order to make a +direct method call on that sub-system. + +With the model described in this document, the RPC server just needs to know +about what is effectively an _abstract address_ of that sub-system. It can then +use that to obtain something similar to a mailbox to do the method call. + +This allows for a more decoupled architecture, as the RPC server doesn't need to +know the exact "shape" of the method to call, just which message to send. +Refactors of the sub-system won't break the RPC server, as long as the message +(which can be constructed via a dedicated constructor) is the same. + +--- + +This package provides a foundational actor framework tailored for Go, enabling +developers to build components that are easier to reason about, test, and +maintain in a concurrent environment. + +## Core Concepts + +Let's explore the fundamental building blocks provided by this package. + +### Messages + +Actors communicate by sending and receiving messages. Any type that an actor +needs to process must implement the `actor.Message` interface. A simple way to +do this is by embedding `actor.BaseMessage`: + +```go +package mymodule + +import "github.com/lightningnetwork/lnd/actor" + +// MyRequest is a custom message type. +type MyRequest struct { + // Embed BaseMessage to satisfy the Message interface. + actor.BaseMessage + Data string +} + +// MessageType returns a string identifier for this message type. +func (m *MyRequest) MessageType() string { + return "MyRequest" +} + +// MyResponse might be a corresponding response type. +type MyResponse struct { + actor.BaseMessage + Reply string +} + +func (m *MyResponse) MessageType() string { + return "MyResponse" +} +``` +The `MessageType()` method provides a string representation of the message type, +which can be useful for debugging or routing. + + +### Actor Behavior + +The logic of an actor (how it responds to messages) is defined by its +`ActorBehavior`. This is an interface that you implement: + +```go +package actor + +// ActorBehavior defines the logic for how an actor processes incoming messages. +type ActorBehavior[M Message, R any] interface { + Receive(actorCtx context.Context, msg M) fn.Result[R] +} +``` +The `Receive` method passes in a caller context (useful for shutdown detection) +and the incoming message. It returns an `fn.Result[R]`, which can encapsulate +either a successful response of type `R` or an error. + +For simple cases, you can use `actor.FunctionBehavior` to adapt a Go function +into an `ActorBehavior`: + +```go +import ( + "context" + "fmt" + "github.com/lightningnetwork/lnd/actor" + "github.com/lightningnetwork/lnd/fn/v2" +) + +// myActorLogic defines the processing for MyRequest messages. +func myActorLogic(ctx context.Context, msg *MyRequest) fn.Result[*MyResponse] { + // In a real actor, you might interact with state or other services. + // The actor's context (ctx) can be checked for shutdown signals. + select { + case <-ctx.Done(): + return fn.Err[*MyResponse](errors.New("actor shutting down")) + default: + } + + response := &MyResponse{Reply: fmt.Sprintf("Processed: %s", msg.Data)} + return fn.Ok(response) +} + +// Create a behavior from the function. +behavior := actor.NewFunctionBehavior(myActorLogic) +``` + +For more complex cases, you can implement the `Receive` method on a new struct, +and pass that around directly. + +### Service Keys and Actor References: The Interaction Layer + +Direct interaction with an actor's internal state or its concrete struct is +discouraged. Instead, communication and discovery are managed through two key +abstractions: `ServiceKey` and `ActorRef`. These provide a layer of indirection, +promoting loose coupling and location transparency (though the current +implementation is in-process). + +#### `ServiceKey[M Message, R any]` + +A `ServiceKey` is a type-safe identifier used for registering actors that +provide a particular service and for discovering them later. The generic type +parameters `M` (the type of message the actor handles) and `R` (the type of +response the actor produces for `Ask` operations) ensure that you discover +actors compatible with the interactions you intend to perform. + +```go +// Define a service key for actors that handle MyRequest and produce MyResponse. +myServiceKey := actor.NewServiceKey[*MyRequest, *MyResponse]("my-custom-service") + +// Later, this key would be used with a Receptionist (part of an ActorSystem) +// to find ActorRefs for actors offering this service. +``` + +#### `ActorRef[M Message, R any]` + +An `ActorRef` is a lightweight, shareable reference to an actor. It's the +primary means by which you send messages to an actor. It is also generic over +the message type `M` and response type `R` that the target actor handles. + +You typically obtain an `ActorRef` by looking it up in a `Receptionist` using a +`ServiceKey` (covered later when discussing the `ActorSystem`), or directly from +an actor instance via its `.Ref()` method (e.g., `sampleActor.Ref()` if you have +the `Actor` instance). + +There are two main ways to send messages using an `ActorRef`: + +1. **Tell (Fire-and-Forget)**: Used for sending messages when you don't need a + direct reply. The call returns immediately after attempting to enqueue the + message. + + ```go + // Assuming 'actorRef' is an ActorRef[*MyRequest, *MyResponse] obtained for an actor. + requestMsg := &MyRequest{Data: "A fire-and-forget message"} + actorRef.Tell(context.Background(), requestMsg) + // The message is now in the actor's mailbox (or will be shortly). + ``` + The `context.Context` passed to `Tell` can be used to cancel the send + operation if, for example, the actor's mailbox is full and the send would + block for too long. + +2. **Ask (Request-Response)**: Used when you need a response from the actor. + This returns a `Future[R]`, which represents the eventual reply. + + ```go + // Assuming 'actorRef' is an ActorRef[*MyRequest, *MyResponse]. + askMsg := &MyRequest{Data: "A request needing a response"} + futureResponse := actorRef.Ask(context.Background(), askMsg) + ``` + A `Future[R]` represents a result that will be available at some point. You + can block until it's ready using `Await`: + + ```go + // Await the result. It's good practice to use a context with a timeout. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + result := futureResponse.Await(ctx) + response, err := result.Unpack() + if err != nil { + fmt.Printf("Ask failed: %v\n", err) + // return or handle error + } else { + fmt.Printf("Received reply: %s\n", response.Reply) + } + ``` + The `Future` interface also offers non-blocking ways to handle results, like + `OnComplete` (for callbacks) and `ThenApply` (for chaining transformations). + A more restricted `TellOnlyRef[M]` is also available if only fire-and-forget + semantics are required (obtained via an actor's `TellRef()` method). + +### Actors + +An `Actor` is the concrete entity that runs a behavior, manages a mailbox, and +has a lifecycle. You create an actor using `actor.NewActor` with an +`ActorConfig`: + +```go +cfg := actor.ActorConfig[*MyRequest, *MyResponse]{ + ID: "my-sample-actor", + Behavior: behavior, + MailboxSize: 10, + // Dead Letter Office (covered later) + DLO: nil, +} +sampleActor, err := actor.NewActor(cfg) +if err != nil { + // Handle invalid config (empty ID, nil behavior). + return err +} +``` + +An actor doesn't start processing messages until its `Start()` method is called. +This launches a dedicated goroutine for the actor. + +```go +sampleActor.Start() +``` +To stop an actor, you call its `Stop()` method. This cancels the actor's +internal context, causing its goroutine to clean up and exit. + +```go +// Sometime later... +sampleActor.Stop() +``` + + +## Visualizing Actor Relationships + +The following diagram illustrates the primary components of the actor package +and their relationships. It provides a high-level overview of how actors are +managed, discovered, and interacted with. + +```mermaid +classDiagram + direction TB + + class ActorSystem { + +Receptionist + +DeadLetters + +Shutdown() + } + + class Receptionist { + +Find(ServiceKey) ActorRef[] + +Register(ServiceKey, ActorRef) + } + + class DeadLetterOffice { + +Receive(undeliverable Message) + } + + class ServiceKey { + +Spawn(ActorSystem, Behavior) ActorRef + } + + class Actor { + -mailbox + -behavior + +Ref() ActorRef + +Start() + +Stop() + } + + class ActorRef { + <> + +Tell(Message) + +Ask(Message) Future + } + + class Message { + <> + } + + class Future { + +Await() Result + } + + class Router { + +Tell(Message) + +Ask(Message) Future + } + + %% Core system relationships + ActorSystem *-- Receptionist : has + ActorSystem *-- DeadLetterOffice : provides + ActorSystem o-- "manages" Actor + + %% Actor and communication + Actor --> ActorRef : provides + Actor ..> Message : processes + ActorRef ..> Message : sends + ActorRef ..> Future : returns for Ask + + %% Service discovery and routing + Receptionist o-- ServiceKey : uses for lookup + ServiceKey ..> Actor : creates + Router --> ActorRef : routes to + Router --> Receptionist : discovers actors via + + note for ActorSystem "Central manager for actor lifecycle and service discovery" + note for Actor "Independent unit with encapsulated state and behavior" + note for ActorRef "Location-transparent handle for sending messages" + note for Message "Data exchanged between actors" + note for ServiceKey "Type-safe identifier for actor registration and discovery" + note for Router "Distributes messages among multiple actors" + note for DeadLetterOffice "Handles messages that cannot be delivered" +``` + +## The Actor System + +While individual actors are useful, they often need to be managed and +coordinated. The `ActorSystem` serves this purpose. + +```go +system := actor.NewActorSystem() +// Ensures all actors in the system are stopped. +defer system.Shutdown() +``` + +### Actor Lifecycle and Registration + +The `ActorSystem` can manage the lifecycle of actors. You can register actors +with the system: + +```go +// Using 'behavior' from earlier and 'myServiceKey' defined in the +// "Service Keys and Actor References" section. + +// RegisterWithSystem creates, starts, and registers the actor. +actorRefFromSystem := actor.RegisterWithSystem( + system, "system-managed-actor", myServiceKey, behavior, +) +``` + +Alternatively, a `ServiceKey` itself provides a `Spawn` method for convenience: +```go +actorRefSpawned := myServiceKey.Spawn(system, "spawned-actor", behavior) +``` + +Actors registered with the system are automatically stopped when +`system.Shutdown()` is called. You can also stop and remove individual actors +using `system.StopAndRemoveActor(actorID)`. + +A `ServiceKey` is essentially the mailbox address of an actor. + +### Receptionist: Service Discovery + +Actors often need to find other actors to communicate with. The `Receptionist` +facilitates this. Actors are registered with the receptionist using a +`ServiceKey`, which is type-safe. + +```go +// Get the system's receptionist. +receptionist := system.Receptionist() + +// Find actors registered for a specific service key. +foundRefs := actor.FindInReceptionist(receptionist, myServiceKey) +if len(foundRefs) > 0 { + targetActor := foundRefs[0] + targetActor.Tell(context.Background(), &MyRequest{Data: "Hello from a discoverer!"}) +} else { + fmt.Println("No actors found for service key:", myServiceKey) +} +``` +When an actor is stopped (e.g., via `ServiceKey.Unregister` or system shutdown), +it should also be unregistered from the receptionist. + +### Dead Letter Office (DLO) + +What happens to messages that cannot be delivered? For example, if an actor is +stopped while messages are still in its mailbox, or if a message is sent to an +actor that doesn't exist (though the current `ActorRef` design makes the latter +less likely for direct sends). + +The `ActorSystem` provides a default `DeadLetterActor`. When an actor is +configured (via `ActorConfig.DLO`), undeliverable messages (e.g., those drained +from its mailbox upon shutdown) can be routed to this DLO. This allows for +logging, auditing, or potential manual intervention for "lost" messages. + +```go +// Actors created via RegisterWithSystem or ServiceKey.Spawn +// are automatically configured to use the system's DLO. +// system.DeadLetters() returns an ActorRef to the system's DLO. +``` + +## Routers: Distributing Work + +Sometimes, you might have multiple actors performing the same kind of task, and +you want to distribute messages among them. A `Router` can do this. It's not an +actor itself but acts as a dispatcher. + +A `Router` uses a `RoutingStrategy` to pick one actor from a group registered +under a `ServiceKey`. + +```go +// Assume 'system' and 'myServiceKey' are set up, and multiple actors +// are registered with 'myServiceKey'. + +// Create a round-robin routing strategy. +roundRobinStrategy := actor.NewRoundRobinStrategy[*MyRequest, *MyResponse]() + +// Create a router for 'myServiceKey' using this strategy. +// Messages sent to this router will be forwarded to one of the actors +// registered under 'myServiceKey'. +// The router also needs a DLO for messages it can't route (e.g., if no actors are available). +serviceRouter := actor.NewRouter( + system.Receptionist(), + myServiceKey, + roundRobinStrategy, + system.DeadLetters(), +) + +// Now, interact with the router as if it were an ActorRef: +serviceRouter.Tell(context.Background(), &MyRequest{Data: "Message via router"}) + +futureReplyFromRouter := serviceRouter.Ask(context.Background(), &MyRequest{Data: "Ask via router"}) +// ... await futureReplyFromRouter ... +``` +If the router cannot find any available actors for the `ServiceKey` (e.g., none +are registered or running), `Tell` operations will typically send the message to +the router's configured DLO, and `Ask` operations will return a `Future` +completed with `ErrNoActorsAvailable`.