actor: add Future[T] and Promise[T] w/ concrete impls
What changed, and why it matters
This commit adds a new internal programming helper—similar to a 'delivery receipt' for asynchronous tasks—to the lnd codebase. It does not change any existing feature, fix a bug, or alter how money or network messages are handled. There is nothing in the commit message or code that suggests a security problem or a response to a reported vulnerability.
No security action required. Review as normal code-quality change if desired.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit introduces generic Future[T] and Promise[T] abstractions in a new actor/ package, with a concrete implementation backed by sync.Once, an atomic.Pointer cache, and a done channel. It provides Await, ThenApply, and OnComplete methods, plus idempotent Complete. The change is purely additive (671 lines across three new files) and contains no modifications to existing logic, no privilege changes, no cryptographic operations, and no network-facing behavior.
Changed components
actor/future.goactor/future_test.goactor/interface.goInspect captured patch +671 / −0
diff --git a/actor/future.go b/actor/future.go
new file mode 100644
index 0000000..8c21169
--- /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 0000000..3d56d2c
--- /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 0000000..8a59509
--- /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
+}
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.