refactor: refine coordination service client

- rename OpenSession to CreateSession
- make CreateSession ctx argument define the context of the method
  not the whole session
- make coordination_test tests fail on errors

See also #53

Author: arkhipov

coordination/coordination.go

@@ -16,20 +16,20 @@ type Client interface {
 	DropNode(ctx context.Context, path string) (err error)
 	DescribeNode(ctx context.Context, path string) (_ *scheme.Entry, _ *NodeConfig, err error)
 
-	// OpenSession starts a new session. This method blocks until the server session is created. The context provided is
-	// used for the lifetime of the session.
+	// CreateSession starts a new session. This method blocks until the server session is created. The context provided
+	// may be used to cancel the invocation. If the method completes successfully, the session remains alive even if
+	// the context is canceled.
 	//
 	// To ensure resources are not leaked, one of the following actions must be performed:
 	//
-	// - cancel the provided context,
 	// - call Close on the Session,
 	// - close the Client which the session was created with,
 	// - call any method of the Session until the ErrSessionClosed is returned.
 	//
 	// # Experimental
 	//
 	// Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
-	OpenSession(ctx context.Context, path string, opts ...options.OpenSessionOption) (Session, error)
+	CreateSession(ctx context.Context, path string, opts ...options.CreateSessionOption) (Session, error)
 }
 
 const (
@@ -81,7 +81,8 @@ type Session interface {
 	// UpdateSemaphore changes semaphore data. This method waits until the server successfully updates the semaphore or
 	// returns an error.
 	//
-	// This method is idempotent. The client will automatically retry in the case of network or server failure.
+	// This method is not idempotent. The client will automatically retry in the case of network or server failure
+	// unless it leaves the client state inconsistent.
 	UpdateSemaphore(ctx context.Context, name string, opts ...options.UpdateSemaphoreOption) error
 
 	// DeleteSemaphore deletes an existing semaphore. This method waits until the server successfully deletes the

coordination/example_test.go

@@ -81,14 +81,14 @@ func Example_semaphore() {
 	}
 	fmt.Printf("node description: %+v\nnode config: %+v\n", e, c)
 
-	s, err := db.Coordination().OpenSession(ctx, "/local/test")
+	s, err := db.Coordination().CreateSession(ctx, "/local/test")
 	if err != nil {
-		fmt.Printf("failed to open session: %v\n", err)
+		fmt.Printf("failed to create session: %v\n", err)
 
 		return
 	}
 	defer s.Close(ctx)
-	fmt.Printf("session 1 opened, id: %d\n", s.SessionID())
+	fmt.Printf("session 1 created, id: %d\n", s.SessionID())
 
 	err = s.CreateSemaphore(ctx, "my-semaphore", 20, options.WithCreateData([]byte{1, 2, 3}))
 	if err != nil {

coordination/options/options.go

@@ -7,71 +7,71 @@ import (
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Coordination"
 )
 
-// WithDescription returns an OpenSessionOption that specifies a user-defined description that may be used to describe
+// WithDescription returns an CreateSessionOption that specifies a user-defined description that may be used to describe
 // the client.
-func WithDescription(description string) OpenSessionOption {
-	return func(c *OpenSessionOptions) {
+func WithDescription(description string) CreateSessionOption {
+	return func(c *CreateSessionOptions) {
 		c.Description = description
 	}
 }
 
-// WithSessionTimeout returns an OpenSessionOption that specifies the timeout during which client may restore a
+// WithSessionTimeout returns an CreateSessionOption that specifies the timeout during which client may restore a
 // detached session. The client is forced to terminate the session if the last successful session request occurred
 // earlier than this time.
 //
 // If this is not set, the client uses the default 5 seconds.
-func WithSessionTimeout(timeout time.Duration) OpenSessionOption {
-	return func(c *OpenSessionOptions) {
+func WithSessionTimeout(timeout time.Duration) CreateSessionOption {
+	return func(c *CreateSessionOptions) {
 		c.SessionTimeout = timeout
 	}
 }
 
-// WithSessionStartTimeout returns an OpenSessionOption that specifies the time that the client should wait for a
+// WithSessionStartTimeout returns an CreateSessionOption that specifies the time that the client should wait for a
 // response to the StartSession request from the server before it terminates the gRPC stream and tries to reconnect.
 //
 // If this is not set, the client uses the default time 1 second.
-func WithSessionStartTimeout(timeout time.Duration) OpenSessionOption {
-	return func(c *OpenSessionOptions) {
+func WithSessionStartTimeout(timeout time.Duration) CreateSessionOption {
+	return func(c *CreateSessionOptions) {
 		c.SessionStartTimeout = timeout
 	}
 }
 
-// WithSessionStopTimeout returns an OpenSessionOption that specifies the time that the client should wait for a
+// WithSessionStopTimeout returns an CreateSessionOption that specifies the time that the client should wait for a
 // response to the StopSession request from the server before it terminates the gRPC stream and tries to reconnect.
 //
 // If this is not set, the client uses the default time 1 second.
-func WithSessionStopTimeout(timeout time.Duration) OpenSessionOption {
-	return func(c *OpenSessionOptions) {
+func WithSessionStopTimeout(timeout time.Duration) CreateSessionOption {
+	return func(c *CreateSessionOptions) {
 		c.SessionStartTimeout = timeout
 	}
 }
 
-// WithSessionKeepAliveTimeout returns an OpenSessionOption that specifies the time that the client will wait before it
-// terminates the gRPC stream and tries to reconnect if no successful responses have been received from the server.
+// WithSessionKeepAliveTimeout returns an CreateSessionOption that specifies the time that the client will wait before
+// it terminates the gRPC stream and tries to reconnect if no successful responses have been received from the server.
 //
 // If this is not set, the client uses the default time 10 seconds.
-func WithSessionKeepAliveTimeout(timeout time.Duration) OpenSessionOption {
-	return func(c *OpenSessionOptions) {
+func WithSessionKeepAliveTimeout(timeout time.Duration) CreateSessionOption {
+	return func(c *CreateSessionOptions) {
 		c.SessionKeepAliveTimeout = timeout
 	}
 }
 
-// WithSessionReconnectDelay returns an OpenSessionOption that specifies the time that the client will wait before it
+// WithSessionReconnectDelay returns an CreateSessionOption that specifies the time that the client will wait before it
 // tries to reconnect the underlying gRPC stream in case of error.
 //
 // If this is not set, the client uses the default time 500 milliseconds.
-func WithSessionReconnectDelay(delay time.Duration) OpenSessionOption {
-	return func(c *OpenSessionOptions) {
+func WithSessionReconnectDelay(delay time.Duration) CreateSessionOption {
+	return func(c *CreateSessionOptions) {
 		c.SessionReconnectDelay = delay
 	}
 }
 
-// OpenSessionOption configures how we open a new session.
-type OpenSessionOption func(c *OpenSessionOptions)
+// CreateSessionOption configures how we create a new session.
+type CreateSessionOption func(c *CreateSessionOptions)
 
-// OpenSessionOptions configure an OpenSession call. OpenSessionOptions are set by the OpenSessionOption values passed
-// to the OpenSession function.
-type OpenSessionOptions struct {
+// CreateSessionOptions configure an CreateSession call. CreateSessionOptions are set by the CreateSessionOption values
+// passed to the CreateSession function.
+type CreateSessionOptions struct {
 	Description             string
 	SessionTimeout          time.Duration
 	SessionStartTimeout     time.Duration
@@ -95,18 +95,18 @@ func WithEphemeral(ephemeral bool) AcquireSemaphoreOption {
 // acquired by another session.
 //
 // If this is not set, the client waits for the acquire operation result until the operation or session context is done.
-// You can reset the default value of this timeout by calling the WithNoAcquireTimeout method.
+// You can reset the default value of this timeout by calling the WithAcquireInfiniteTimeout method.
 func WithAcquireTimeout(timeout time.Duration) AcquireSemaphoreOption {
 	return func(c *Ydb_Coordination.SessionRequest_AcquireSemaphore) {
 		c.TimeoutMillis = uint64(timeout.Milliseconds())
 	}
 }
 
-// WithNoAcquireTimeout returns an AcquireSemaphoreOption which disables the timeout after which the operation fails if
-// it is still waiting in the queue.
+// WithAcquireInfiniteTimeout returns an AcquireSemaphoreOption which disables the timeout after which the operation
+// fails if it is still waiting in the queue.
 //
 // This is the default behavior. You can set the specific timeout by calling the WithAcquireTimeout method.
-func WithNoAcquireTimeout() AcquireSemaphoreOption {
+func WithAcquireInfiniteTimeout() AcquireSemaphoreOption {
 	return func(c *Ydb_Coordination.SessionRequest_AcquireSemaphore) {
 		c.TimeoutMillis = math.MaxUint64
 	}

examples/coordination/lock/main.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"os/signal"
 	"sync"
+	"syscall"
 	"time"
 
 	environ "github.com/ydb-platform/ydb-go-sdk-auth-environ"
@@ -62,7 +63,7 @@ func init() {
 }
 
 func main() {
-	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer cancel()
 
 	db, err := ydb.Open(ctx, dsn,
@@ -90,7 +91,7 @@ func main() {
 	for {
 		fmt.Println("waiting for a lock...")
 
-		session, err := db.Coordination().OpenSession(ctx, path)
+		session, err := db.Coordination().CreateSession(ctx, path)
 		if err != nil {
 			fmt.Println("failed to open session", err)
 

examples/coordination/workers/main.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"os/signal"
 	"sync"
+	"syscall"
 	"time"
 
 	environ "github.com/ydb-platform/ydb-go-sdk-auth-environ"
@@ -74,13 +75,8 @@ func init() {
 }
 
 func main() {
-	ctx, cancel := context.WithCancel(context.Background())
-	c := make(chan os.Signal, 1)
-	signal.Notify(c, os.Interrupt)
-	defer func() {
-		signal.Stop(c)
-		cancel()
-	}()
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
 
 	db, err := ydb.Open(ctx, dsn,
 		environ.WithEnvironCredentials(ctx),
@@ -114,7 +110,7 @@ func main() {
 
 	fmt.Println("starting tasks")
 	for {
-		session, err := db.Coordination().OpenSession(ctx, path)
+		session, err := db.Coordination().CreateSession(ctx, path)
 		if err != nil {
 			fmt.Println("failed to open session", err)
 
@@ -137,16 +133,16 @@ func main() {
 			select {
 			case <-semaphoreCtx.Done():
 				break loop
-			case <-c:
-				fmt.Println("exiting")
-
-				return
 			case lease := <-leaseChan:
 				go doWork(lease.lease, lease.semaphoreName)
 				tasksStarted++
 				if tasksStarted == capacity {
 					break loop
 				}
+			case <-ctx.Done():
+				fmt.Println("exiting")
+
+				return
 			}
 		}
 
@@ -156,11 +152,11 @@ func main() {
 		wg.Wait()
 
 		select {
-		case <-session.Context().Done():
-		case <-c:
+		case <-ctx.Done():
 			fmt.Println("exiting")
 
 			return
+		case <-session.Context().Done():
 		}
 	}
 }

internal/coordination/client.go

@@ -249,8 +249,8 @@ func (c *Client) describeNode(
 	}, nil
 }
 
-func newOpenSessionConfig(opts ...options.OpenSessionOption) *options.OpenSessionOptions {
-	c := defaultOpenSessionConfig()
+func newCreateSessionConfig(opts ...options.CreateSessionOption) *options.CreateSessionOptions {
+	c := defaultCreateSessionConfig()
 	for _, o := range opts {
 		if o != nil {
 			o(c)
@@ -260,7 +260,7 @@ func newOpenSessionConfig(opts ...options.OpenSessionOption) *options.OpenSessio
 	return c
 }
 
-func (c *Client) sessionOpened(s *session) {
+func (c *Client) sessionCreated(s *session) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
@@ -283,8 +283,8 @@ func (c *Client) closeSessions(ctx context.Context) {
 	}
 }
 
-func defaultOpenSessionConfig() *options.OpenSessionOptions {
-	return &options.OpenSessionOptions{
+func defaultCreateSessionConfig() *options.CreateSessionOptions {
+	return &options.CreateSessionOptions{
 		Description:             "YDB Go SDK",
 		SessionTimeout:          time.Second * 5,
 		SessionStartTimeout:     time.Second * 1,
@@ -294,16 +294,16 @@ func defaultOpenSessionConfig() *options.OpenSessionOptions {
 	}
 }
 
-func (c *Client) OpenSession(
+func (c *Client) CreateSession(
 	ctx context.Context,
 	path string,
-	opts ...options.OpenSessionOption,
+	opts ...options.CreateSessionOption,
 ) (coordination.Session, error) {
 	if c == nil {
 		return nil, xerrors.WithStackTrace(errNilClient)
 	}
 
-	return openSession(ctx, c, path, newOpenSessionConfig(opts...))
+	return createSession(ctx, c, path, newCreateSessionConfig(opts...))
 }
 
 func (c *Client) Close(ctx context.Context) error {