Create lifecycle state

Author: exanubes

connect.go

@@ -9,6 +9,7 @@ import (
 	"github.com/exanubes/appsync/internal/app"
 	"github.com/exanubes/appsync/internal/app/engine"
 	"github.com/exanubes/appsync/internal/app/heartbeat"
+	"github.com/exanubes/appsync/internal/app/lifecycle"
 	"github.com/exanubes/appsync/internal/app/pending"
 	"github.com/exanubes/appsync/internal/app/queue"
 	"github.com/exanubes/appsync/internal/app/router"
@@ -133,6 +134,7 @@ func (builder *builder) Connect(ctx context.Context) (*appsync_client, error) {
 		return nil, err
 	}
 
+	connection_state := lifecycle.NewState()
 	clock := clock.New()
 	heartbeat := heartbeat.New(clock)
 	ingress_queue := queue.NewIngressQueue(builder.backpressure.ConnectionInbound)
@@ -149,7 +151,13 @@ func (builder *builder) Connect(ctx context.Context) (*appsync_client, error) {
 
 	msg_router := router.New(pending_registry, usecases.ReceiveData)
 	runtime := runtime.New(ingress_queue, msg_router, heartbeat)
-	session := engine.New(heartbeat, runtime, io_loops, builder.logger)
+	session := engine.New(
+		heartbeat,
+		runtime,
+		io_loops,
+		connection_state,
+		builder.logger,
+	)
 	session.Start(ctx, engine.StartEngineInput{
 		Timeout: connection_output.Timeout,
 	})

errors.go

@@ -10,4 +10,5 @@ var (
 	ErrSubscriptionClosed    = app.ErrSubscriptionClosed
 	ErrSubscriptionNotFound  = app.ErrSubscriptionNotFound
 	ErrHeartbeatTimeout      = app.ErrHeartbeatTimeout
+	ErrConnectionClosed      = app.ErrConnectionClosed
 )

internal/app/engine/engine.go

@@ -17,16 +17,18 @@ type Engine struct {
 	wg          sync.WaitGroup
 	ctx         context.Context
 	cancel      context.CancelFunc
+	state       ConnectionState
 }
 
 var managed_goroutines_count = 4
 
-func New(heartbeat app.Heartbeat, runtime Runtime, io IO, logger app.Logger) *Engine {
+func New(heartbeat app.Heartbeat, runtime Runtime, io IO, state ConnectionState, logger app.Logger) *Engine {
 	return &Engine{
 		io:          io,
 		logger:      logger.SetContext("Engine"),
 		runtime:     runtime,
 		heartbeat:   heartbeat,
+		state:       state,
 		err_channel: make(chan error, managed_goroutines_count),
 	}
 }
@@ -35,28 +37,36 @@ func (engine *Engine) Start(ctx context.Context, input StartEngineInput) {
 	engine.ctx, engine.cancel = context.WithCancel(ctx)
 	engine.wg.Add(managed_goroutines_count)
 	go func() {
-		engine.err_channel <- engine.io.Read(engine.ctx)
+		err := engine.io.Read(engine.ctx)
+		engine.state.Close(err)
+		engine.err_channel <- err
 		engine.logger.Debug("Exitted ingress loop")
 		engine.wg.Done()
 		engine.cancel()
 	}()
 
 	go func() {
-		engine.err_channel <- engine.io.Write(engine.ctx)
+		err := engine.io.Write(engine.ctx)
+		engine.state.Close(err)
+		engine.err_channel <- err
 		engine.logger.Debug("Exitted egress loop")
 		engine.wg.Done()
 		engine.cancel()
 	}()
 
 	go func() {
-		engine.err_channel <- engine.runtime.Run(engine.ctx)
+		err := engine.runtime.Run(engine.ctx)
+		engine.state.Close(err)
+		engine.err_channel <- err
 		engine.logger.Debug("Exitted runtime loop")
 		engine.wg.Done()
 		engine.cancel()
 	}()
 
 	go func() {
-		engine.err_channel <- engine.heartbeat.Start(engine.ctx, input.Timeout)
+		err := engine.heartbeat.Start(engine.ctx, input.Timeout)
+		engine.state.Close(err)
+		engine.err_channel <- err
 		engine.logger.Debug("Exitted heartbeat loop")
 		engine.wg.Done()
 		engine.cancel()

internal/app/engine/engine_test.go

@@ -68,9 +68,13 @@ func (m *mock_heartbeat) Start(ctx context.Context, _ time.Duration) error {
 
 func (m *mock_heartbeat) Reset() {}
 
+type mock_connection_state struct{}
+
+func (m *mock_connection_state) Close(cause error) {}
+
 func TestNew(t *testing.T) {
 	logger := &stub_logger{}
-	engine.New(&mock_heartbeat{}, &mock_runtime{}, &mock_io{}, logger)
+	engine.New(&mock_heartbeat{}, &mock_runtime{}, &mock_io{}, &mock_connection_state{}, logger)
 	if logger.ctx != "Engine" {
 		t.Errorf("expected logger context 'Engine', got %q", logger.ctx)
 	}
@@ -81,7 +85,7 @@ func TestStart_calls_all_goroutines(t *testing.T) {
 	runtime := &mock_runtime{}
 	hb := &mock_heartbeat{}
 
-	e := engine.New(hb, runtime, io, &stub_logger{})
+	e := engine.New(hb, runtime, io, &mock_connection_state{}, &stub_logger{})
 	e.Start(context.Background(), engine.StartEngineInput{Timeout: time.Second})
 	e.Close(context.Background())
 
@@ -137,7 +141,7 @@ func TestClose(t *testing.T) {
 			runtime := &mock_runtime{err: tt.err}
 			hb := &mock_heartbeat{err: tt.err}
 
-			e := engine.New(hb, runtime, io, &stub_logger{})
+			e := engine.New(hb, runtime, io, &mock_connection_state{}, &stub_logger{})
 			e.Start(context.Background(), engine.StartEngineInput{Timeout: time.Second})
 			err := e.Close(context.Background())
 

internal/app/engine/ports.go

@@ -31,3 +31,7 @@ type Runtime interface {
 type Inbox interface {
 	Next(context.Context) (app.Message, error)
 }
+
+type ConnectionState interface {
+	Close(error)
+}

internal/app/errors.go

@@ -9,3 +9,4 @@ var ErrSubscriptionInboxFull = errors.New("Subscription incoming message buffer
 var ErrSubscriptionClosed = errors.New("Subscription is closed")
 var ErrSubscriptionNotFound = errors.New("Subscription not found")
 var ErrHeartbeatTimeout = errors.New("Heartbeat timeout")
+var ErrConnectionClosed = errors.New("Connection closed")

internal/app/lifecycle/lifecycle.go

@@ -0,0 +1,59 @@
+package lifecycle
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/exanubes/appsync/internal/app"
+)
+
+type State struct {
+	once  sync.Once
+	done  chan struct{}
+	mutex sync.RWMutex
+	err   error
+}
+
+func NewState() *State {
+	return &State{
+		done: make(chan struct{}),
+	}
+}
+
+func (state *State) Done() <-chan struct{} {
+	return state.done
+}
+
+func (state *State) Err() error {
+	select {
+	case <-state.done:
+		state.mutex.RLock()
+		defer state.mutex.RUnlock()
+
+		if state.err == nil {
+			return app.ErrConnectionClosed
+		}
+
+		return state.err
+	default:
+		return nil
+	}
+}
+
+func (state *State) Close(cause error) {
+	state.once.Do(func() {
+		if cause == nil {
+			cause = app.ErrConnectionClosed
+		}
+
+		if !errors.Is(cause, app.ErrConnectionClosed) {
+			cause = errors.Join(app.ErrConnectionClosed, cause)
+		}
+
+		state.mutex.Lock()
+		state.err = cause
+		state.mutex.Unlock()
+
+		close(state.done)
+	})
+}

internal/app/lifecycle/lifecycle_test.go

@@ -0,0 +1,159 @@
+package lifecycle_test
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/exanubes/appsync/internal/app"
+	"github.com/exanubes/appsync/internal/app/lifecycle"
+)
+
+var errCustom = errors.New("custom error")
+
+func closed_state(cause error) *lifecycle.State {
+	s := lifecycle.NewState()
+	s.Close(cause)
+	return s
+}
+
+func TestNewState(t *testing.T) {
+	t.Run("done channel is open", func(t *testing.T) {
+		s := lifecycle.NewState()
+		select {
+		case <-s.Done():
+			t.Fatal("Done() should not be readable on a new state")
+		default:
+		}
+	})
+}
+
+func TestDone(t *testing.T) {
+	t.Run("blocks before Close", func(t *testing.T) {
+		s := lifecycle.NewState()
+		select {
+		case <-s.Done():
+			t.Fatal("Done() should block before Close()")
+		default:
+		}
+	})
+
+	t.Run("readable after Close", func(t *testing.T) {
+		s := closed_state(nil)
+		select {
+		case <-s.Done():
+		case <-time.After(100 * time.Millisecond):
+			t.Fatal("Done() should be readable after Close()")
+		}
+	})
+}
+
+func TestErr(t *testing.T) {
+	tests := []struct {
+		name      string
+		state     func() *lifecycle.State
+		is_closed bool
+		is_custom bool
+	}{
+		{
+			name:      "returns nil before Close",
+			state:     lifecycle.NewState,
+			is_closed: false,
+			is_custom: false,
+		},
+		{
+			name:      "returns ErrConnectionClosed when closed with nil",
+			state:     func() *lifecycle.State { return closed_state(nil) },
+			is_closed: true,
+			is_custom: false,
+		},
+		{
+			name:      "returns ErrConnectionClosed when closed with ErrConnectionClosed",
+			state:     func() *lifecycle.State { return closed_state(app.ErrConnectionClosed) },
+			is_closed: true,
+			is_custom: false,
+		},
+		{
+			name:      "wraps custom error with ErrConnectionClosed",
+			state:     func() *lifecycle.State { return closed_state(errCustom) },
+			is_closed: true,
+			is_custom: true,
+		},
+		{
+			name:      "preserves pre-joined error containing ErrConnectionClosed and custom error",
+			state:     func() *lifecycle.State { return closed_state(errors.Join(app.ErrConnectionClosed, errCustom)) },
+			is_closed: true,
+			is_custom: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.state().Err()
+
+			if !tt.is_closed {
+				if err != nil {
+					t.Errorf("Err() = %v, want nil", err)
+				}
+				return
+			}
+
+			if err == nil {
+				t.Fatal("Err() = nil, want non-nil error")
+			}
+			if !errors.Is(err, app.ErrConnectionClosed) {
+				t.Errorf("Err() = %v, want errors.Is(err, ErrConnectionClosed) == true", err)
+			}
+			if tt.is_custom && !errors.Is(err, errCustom) {
+				t.Errorf("Err() = %v, want errors.Is(err, errCustom) == true", err)
+			}
+		})
+	}
+}
+
+func TestClose(t *testing.T) {
+	t.Run("closes Done channel", func(t *testing.T) {
+		s := lifecycle.NewState()
+		s.Close(nil)
+		select {
+		case <-s.Done():
+		case <-time.After(100 * time.Millisecond):
+			t.Fatal("Done() should be closed after Close()")
+		}
+	})
+
+	t.Run("idempotent: second call does not overwrite first", func(t *testing.T) {
+		s := lifecycle.NewState()
+		s.Close(errCustom)
+		s.Close(nil)
+
+		if err := s.Err(); !errors.Is(err, errCustom) {
+			t.Errorf("Err() = %v, want errors.Is(err, errCustom) == true after second Close(nil)", err)
+		}
+	})
+
+	t.Run("concurrent calls are safe and close exactly once", func(t *testing.T) {
+		const goroutines = 100
+		s := lifecycle.NewState()
+
+		var wg sync.WaitGroup
+		wg.Add(goroutines)
+		start := make(chan struct{})
+
+		for range goroutines {
+			go func() {
+				defer wg.Done()
+				<-start
+				s.Close(errCustom)
+			}()
+		}
+
+		close(start)
+		wg.Wait()
+
+		if err := s.Err(); err == nil {
+			t.Fatal("Err() = nil after concurrent Close calls, want non-nil")
+		}
+	})
+}

internal/app/services/connection/create_connection_test.go

@@ -146,9 +146,9 @@ func TestConnect(t *testing.T) {
 			},
 		},
 		{
-			name:      "generated subprotocol appended to input subprotocols",
-			generator: &mock_subprotocol_generator{result: "header-xyz"},
-			dialer:    &mock_dialer{conn: conn},
+			name:       "generated subprotocol appended to input subprotocols",
+			generator:  &mock_subprotocol_generator{result: "header-xyz"},
+			dialer:     &mock_dialer{conn: conn},
 			authorizer: &mock_connection_authorizer{timeout: timeout},
 			input: connection.CreateConnectionInput{
 				Url:          endpoint,