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

@@ -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)
+	}()
+}

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",
+		)
+	})
+}

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
+}