Go: introduce context for Transact/ReadTransact

Use a goroutine to automatically cancel transaction when context has an error.
Correctly close transactions by returning a function to caller.

Author: gm42

bindings/go/src/fdb/database.go

@@ -27,6 +27,7 @@ package fdb
 import "C"
 
 import (
+	"context"
 	"errors"
 )
 
@@ -162,23 +163,52 @@ func (d Database) GetClientStatus() ([]byte, error) {
 	return b, nil
 }
 
-func retryable(wrapped func() (interface{}, error), onError func(Error) FutureNil) (ret interface{}, err error) {
+func autoCancel(ctx context.Context, tr CancellableTransaction) chan<- struct{} {
+	exitCh := make(chan struct{})
+	// This goroutine ensures that as soon as context is canceled the FoundationDB transaction will be canceled as well;
+	// this can in turn cause FoundationDB's `transaction_canceled` errors.
+	go func() {
+		select {
+		case <-exitCh:
+			// transaction finished before any context cancellation was acknowledged
+			return
+		case <-ctx.Done():
+			// context cancellation happened before transaction finished
+			// NOTE: canceling the transaction does not mean that transaction state changes did not happen,
+
+			tr.Cancel()
+
+			return
+		}
+	}()
+
+	return exitCh
+}
+
+// retryable is responsible for retrying the wrapped function for as long as the returned error is retryable.
+// In case of a non-retryable error, it is returned and transaction is closed.
+func retryable(ctx context.Context, tr Transaction, wrapped func() (interface{}, error)) (ret interface{}, txClose func(), err error) {
+	// NOTE: 'autoCancel' must be called outside of the defer function, so that the goroutine is started
+	ch := autoCancel(ctx, tr)
+	defer close(ch)
+
 	for {
 		ret, err = wrapped()
-
-		// No error means success!
 		if err == nil {
+			// No error means success!
+			// Caller is responsible for closing transaction after finishing using any future created within the transaction
+			// This return value is always nil in case of errors
+			txClose = tr.Close
+
 			return
 		}
 
-		// Check if the error chain contains an
-		// fdb.Error
+		// Check if the error chain contains an fdb.Error
 		var ep Error
 		if errors.As(err, &ep) {
-			f := onError(ep)
-			defer f.Close()
-
+			f := tr.OnError(ep)
 			processedErr := f.Get()
+			f.Close()
 			var newEp Error
 			if !errors.As(processedErr, &newEp) || newEp.Code != ep.Code {
 				// override original error only if not an Error or code changed
@@ -190,6 +220,10 @@ func retryable(wrapped func() (interface{}, error), onError func(Error) FutureNi
 		// If OnError returns an error, then it's not
 		// retryable; otherwise take another pass at things
 		if err != nil {
+
+			// destroy transaction; this will cancel all futures created within it
+			tr.Close()
+
 			return
 		}
 	}
@@ -207,40 +241,37 @@ func retryable(wrapped func() (interface{}, error), onError func(Error) FutureNi
 // error.
 //
 // The transaction is retried if the error is or wraps a retryable Error.
-// The error is unwrapped.
 //
-// Do not return Future objects from the function provided to Transact. The
-// Transaction created by Transact may be finalized at any point after Transact
-// returns, resulting in the cancellation of any outstanding
-// reads. Additionally, any errors returned or panicked by the Future will no
-// longer be able to trigger a retry of the caller-provided function.
+// In case of success a transaction close function is returned and caller must
+// call it after all futures have been used.
+// Any errors returned or panicked by the Future outside of the Transact() call
+// will no longer be able to trigger a retry of the caller-provided function.
 //
 // See the Transactor interface for an example of using Transact with
 // Transaction and Database objects.
-func (d Database) Transact(f func(Transaction) (interface{}, error)) (interface{}, error) {
+func (d Database) Transact(ctx context.Context, f func(Transaction) (interface{}, error)) (interface{}, func(), error) {
 	tr, err := d.CreateTransaction()
 	if err != nil {
 		// Any error here is non-retryable
-		return nil, err
+		return nil, nil, err
 	}
-	defer tr.Close()
 
 	wrapped := func() (ret interface{}, err error) {
 		defer panicToError(&err)
 
 		ret, err = f(tr)
-
-		if err == nil {
-			f := tr.Commit()
-			defer f.Close()
-
-			err = f.Get()
+		if err != nil {
+			return
 		}
 
+		f := tr.Commit()
+		err = f.Get()
+		f.Close()
+
 		return
 	}
 
-	return retryable(wrapped, tr.OnError)
+	return retryable(ctx, tr, wrapped)
 }
 
 // ReadTransact runs a caller-provided function inside a retry loop, providing
@@ -254,37 +285,33 @@ func (d Database) Transact(f func(Transaction) (interface{}, error)) (interface{
 // transaction or return the error.
 //
 // The transaction is retried if the error is or wraps a retryable Error.
-// The error is unwrapped.
-// Read transactions are never committed and destroyed automatically via GC,
-// once all their futures go out of scope.
+// Read transactions are never committed.
 //
-// Do not return Future objects from the function provided to ReadTransact. The
-// Transaction created by ReadTransact may be finalized at any point after
-// ReadTransact returns, resulting in the cancellation of any outstanding
-// reads. Additionally, any errors returned or panicked by the Future will no
-// longer be able to trigger a retry of the caller-provided function.
+// In case of success a transaction close function is returned and caller must
+// call it after all futures have been used.
+// Any errors returned or panicked by the Future outside of the Transact() call
+// will no longer be able to trigger a retry of the caller-provided function.
 //
 // See the ReadTransactor interface for an example of using ReadTransact with
 // Transaction, Snapshot and Database objects.
-func (d Database) ReadTransact(f func(ReadTransaction) (interface{}, error)) (interface{}, error) {
+func (d Database) ReadTransact(ctx context.Context, f func(ReadTransaction) (interface{}, error)) (interface{}, func(), error) {
 	tr, err := d.CreateTransaction()
 	if err != nil {
 		// Any error here is non-retryable
-		return nil, err
+		return nil, nil, err
 	}
-	defer tr.Close()
 
 	wrapped := func() (ret interface{}, err error) {
 		defer panicToError(&err)
 
 		ret, err = f(tr)
 
-		// read-only transactions are not committed and will be destroyed automatically by the deferred Close()
+		// read-only transactions are not committed and will be destroyed automatically when Transaction's Close() is called
 
 		return
 	}
 
-	return retryable(wrapped, tr.OnError)
+	return retryable(ctx, tr, wrapped)
 }
 
 // Options returns a DatabaseOptions instance suitable for setting options

bindings/go/src/fdb/directory/directory.go

@@ -40,6 +40,7 @@
 package directory
 
 import (
+	"context"
 	"errors"
 
 	"github.com/apple/foundationdb/bindings/go/src/fdb"
@@ -79,7 +80,7 @@ type Directory interface {
 	// recorded as the layer; if layer is specified and the directory already
 	// exists, it is compared against the layer specified when the directory was
 	// created, and an error is returned if they differ.
-	CreateOrOpen(t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error)
+	CreateOrOpen(ctx context.Context, t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error)
 
 	// Open opens the directory specified by path (relative to this Directory),
 	// and returns the directory and its contents as a DirectorySubspace (or ErrDirNotExists
@@ -89,15 +90,15 @@ type Directory interface {
 	// If the byte slice layer is specified, it is compared against the layer
 	// specified when the directory was created, and an error is returned if
 	// they differ.
-	Open(rt fdb.ReadTransactor, path []string, layer []byte) (DirectorySubspace, error)
+	Open(ctx context.Context, rt fdb.ReadTransactor, path []string, layer []byte) (DirectorySubspace, error)
 
 	// Create creates a directory specified by path (relative to this
 	// Directory), and returns the directory and its contents as a
 	// DirectorySubspace (or ErrDirAlreadyExists if the directory already exists).
 	//
 	// If the byte slice layer is specified, it is recorded as the layer and
 	// will be checked when opening the directory in the future.
-	Create(t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error)
+	Create(ctx context.Context, t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error)
 
 	// CreatePrefix behaves like Create, but uses a manually specified byte
 	// slice prefix to physically store the contents of this directory, rather
@@ -106,7 +107,7 @@ type Directory interface {
 	// If this Directory was created in a root directory that does not allow
 	// manual prefixes, CreatePrefix will return an error. The default root
 	// directory does not allow manual prefixes.
-	CreatePrefix(t fdb.Transactor, path []string, layer []byte, prefix []byte) (DirectorySubspace, error)
+	CreatePrefix(ctx context.Context, t fdb.Transactor, path []string, layer []byte, prefix []byte) (DirectorySubspace, error)
 
 	// Move moves the directory at oldPath to newPath (both relative to this
 	// Directory), and returns the directory (at its new location) and its
@@ -116,7 +117,7 @@ type Directory interface {
 	//
 	// There is no effect on the physical prefix of the given directory or on
 	// clients that already have the directory open.
-	Move(t fdb.Transactor, oldPath []string, newPath []string) (DirectorySubspace, error)
+	Move(ctx context.Context, t fdb.Transactor, oldPath []string, newPath []string) (DirectorySubspace, error)
 
 	// MoveTo moves this directory to newAbsolutePath (relative to the root
 	// directory of this Directory), and returns the directory (at its new
@@ -126,7 +127,7 @@ type Directory interface {
 	//
 	// There is no effect on the physical prefix of the given directory or on
 	// clients that already have the directory open.
-	MoveTo(t fdb.Transactor, newAbsolutePath []string) (DirectorySubspace, error)
+	MoveTo(ctx context.Context, t fdb.Transactor, newAbsolutePath []string) (DirectorySubspace, error)
 
 	// Remove removes the directory at path (relative to this Directory), its
 	// content, and all subdirectories. Remove returns true if a directory
@@ -135,16 +136,16 @@ type Directory interface {
 	//
 	// Note that clients that have already opened this directory might still
 	// insert data into its contents after removal.
-	Remove(t fdb.Transactor, path []string) (bool, error)
+	Remove(ctx context.Context, t fdb.Transactor, path []string) (bool, error)
 
 	// Exists returns true if the directory at path (relative to this Directory)
 	// exists, and false otherwise.
-	Exists(rt fdb.ReadTransactor, path []string) (bool, error)
+	Exists(ctx context.Context, rt fdb.ReadTransactor, path []string) (bool, error)
 
 	// List returns the names of the immediate subdirectories of the directory
 	// at path (relative to this Directory) as a slice of strings. Each string
 	// is the name of the last component of a subdirectory's path.
-	List(rt fdb.ReadTransactor, path []string) ([]string, error)
+	List(ctx context.Context, rt fdb.ReadTransactor, path []string) ([]string, error)
 
 	// GetLayer returns the layer specified when this Directory was created.
 	GetLayer() []byte
@@ -165,14 +166,14 @@ func stringsEqual(a, b []string) bool {
 	return true
 }
 
-func moveTo(t fdb.Transactor, dl directoryLayer, path, newAbsolutePath []string) (DirectorySubspace, error) {
+func moveTo(ctx context.Context, t fdb.Transactor, dl directoryLayer, path, newAbsolutePath []string) (DirectorySubspace, error) {
 	partition_len := len(dl.path)
 
 	if !stringsEqual(newAbsolutePath[:partition_len], dl.path) {
 		return nil, errors.New("cannot move between partitions")
 	}
 
-	return dl.Move(t, path[partition_len:], newAbsolutePath[partition_len:])
+	return dl.Move(ctx, t, path[partition_len:], newAbsolutePath[partition_len:])
 }
 
 var root = NewDirectoryLayer(subspace.FromBytes([]byte{0xFE}), subspace.AllKeys(), false)
@@ -186,8 +187,8 @@ var root = NewDirectoryLayer(subspace.FromBytes([]byte{0xFE}), subspace.AllKeys(
 // as the layer; if layer is specified and the directory already exists, it is
 // compared against the layer specified when the directory was created, and an
 // error is returned if they differ.
-func CreateOrOpen(t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error) {
-	return root.CreateOrOpen(t, path, layer)
+func CreateOrOpen(ctx context.Context, t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error) {
+	return root.CreateOrOpen(ctx, t, path, layer)
 }
 
 // Open opens the directory specified by path (resolved relative to the default
@@ -197,8 +198,8 @@ func CreateOrOpen(t fdb.Transactor, path []string, layer []byte) (DirectorySubsp
 // If the byte slice layer is specified, it is compared against the layer
 // specified when the directory was created, and an error is returned if they
 // differ.
-func Open(rt fdb.ReadTransactor, path []string, layer []byte) (DirectorySubspace, error) {
-	return root.Open(rt, path, layer)
+func Open(ctx context.Context, rt fdb.ReadTransactor, path []string, layer []byte) (DirectorySubspace, error) {
+	return root.Open(ctx, rt, path, layer)
 }
 
 // Create creates a directory specified by path (resolved relative to the
@@ -207,8 +208,8 @@ func Open(rt fdb.ReadTransactor, path []string, layer []byte) (DirectorySubspace
 //
 // If the byte slice layer is specified, it is recorded as the layer and will be
 // checked when opening the directory in the future.
-func Create(t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error) {
-	return root.Create(t, path, layer)
+func Create(ctx context.Context, t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, error) {
+	return root.Create(ctx, t, path, layer)
 }
 
 // Move moves the directory at oldPath to newPath (both resolved relative to the
@@ -219,21 +220,21 @@ func Create(t fdb.Transactor, path []string, layer []byte) (DirectorySubspace, e
 //
 // There is no effect on the physical prefix of the given directory or on
 // clients that already have the directory open.
-func Move(t fdb.Transactor, oldPath []string, newPath []string) (DirectorySubspace, error) {
-	return root.Move(t, oldPath, newPath)
+func Move(ctx context.Context, t fdb.Transactor, oldPath []string, newPath []string) (DirectorySubspace, error) {
+	return root.Move(ctx, t, oldPath, newPath)
 }
 
 // Exists returns true if the directory at path (relative to the default root
 // directory) exists, and false otherwise.
-func Exists(rt fdb.ReadTransactor, path []string) (bool, error) {
-	return root.Exists(rt, path)
+func Exists(ctx context.Context, rt fdb.ReadTransactor, path []string) (bool, error) {
+	return root.Exists(ctx, rt, path)
 }
 
 // List returns the names of the immediate subdirectories of the default root
 // directory as a slice of strings. Each string is the name of the last
 // component of a subdirectory's path.
-func List(rt fdb.ReadTransactor, path []string) ([]string, error) {
-	return root.List(rt, path)
+func List(ctx context.Context, rt fdb.ReadTransactor, path []string) ([]string, error) {
+	return root.List(ctx, rt, path)
 }
 
 // Root returns the default root directory. Any attempt to move or remove the