From 3069829d613de68f0bdf56b21c8186dc02debdf1 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Wed, 14 May 2025 18:51:54 -0700 Subject: [PATCH] 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 +}