Repository: ably/ably-go
Pull request #172 (2020-07-23 07:54:15)

add rtn14 - Connection opening failures:
- [x] RTN14a
- [x] RTN14b
- [x] RTN14c
- [x] RTN14d
- [x] RTN14e
- [x] RTN14f
- [x] RTN14g


Patch:
diff --git a/ably/internal/ablyutil/websocket.go b/ably/internal/ablyutil/websocket.go
index 7159372..61aa913 100644
--- a/ably/internal/ablyutil/websocket.go
+++ b/ably/internal/ablyutil/websocket.go
@@ -2,6 +2,7 @@ package ablyutil
 
 import (
 	"errors"
+	"net"
 	"net/url"
 	"time"
 
@@ -38,7 +39,7 @@ func (ws *WebsocketConn) Close() error {
 	return ws.conn.Close()
 }
 
-func DialWebsocket(proto string, u *url.URL) (*WebsocketConn, error) {
+func DialWebsocket(proto string, u *url.URL, timeout time.Duration) (*WebsocketConn, error) {
 	ws := &WebsocketConn{}
 	switch proto {
 	case "application/json":
@@ -48,7 +49,7 @@ func DialWebsocket(proto string, u *url.URL) (*WebsocketConn, error) {
 	default:
 		return nil, errors.New(`invalid protocol "` + proto + `"`)
 	}
-	conn, err := websocket.Dial(u.String(), "", "https://"+u.Host)
+	conn, err := dialWebsocketTimeout(u.String(), "", "https://"+u.Host, timeout)
 	if err != nil {
 		return nil, err
 	}
@@ -56,6 +57,21 @@ func DialWebsocket(proto string, u *url.URL) (*WebsocketConn, error) {
 	return ws, nil
 }
 
+// dialWebsocketTimeout dials the websocket with a timeout.
+func dialWebsocketTimeout(uri, protocol, origin string, timeout time.Duration) (*websocket.Conn, error) {
+	config, err := websocket.NewConfig(uri, origin)
+	if err != nil {
+		return nil, err
+	}
+	if protocol != "" {
+		config.Protocol = []string{protocol}
+	}
+	config.Dialer = &net.Dialer{
+		Timeout: timeout,
+	}
+	return websocket.DialConfig(config)
+}
+
 var msgpackCodec = websocket.Codec{
 	Marshal: func(v interface{}) ([]byte, byte, error) {
 		p, err := MarshalMsgpack(v)
diff --git a/ably/options.go b/ably/options.go
index 53fbcf3..95d459f 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -29,8 +29,11 @@ var defaultOptions = clientOptions{
 	HTTPRequestTimeout:       10 * time.Second,
 	RealtimeHost:             "realtime.ably.io",
 	TimeoutDisconnect:        30 * time.Second,
+	ConnectionStateTTL:       120 * time.Second,
 	RealtimeRequestTimeout:   10 * time.Second, // DF1b
+	SuspendedRetryTimeout:    30 * time.Second, //  RTN14d, TO3l2
 	DisconnectedRetryTimeout: 15 * time.Second, // TO3l1
+	HTTPOpenTimeout:          4 * time.Second,  //TO3l3
 	ChannelRetryTimeout:      15 * time.Second, // TO3l7
 	FallbackRetryTimeout:     10 * time.Minute,
 	IdempotentRESTPublishing: false,
@@ -202,6 +205,8 @@ type clientOptions struct {
 	TimeoutConnect    time.Duration
 	TimeoutDisconnect time.Duration // time period after which disconnect request is failed
 
+	ConnectionStateTTL time.Duration //(DF1a)
+
 	// RealtimeRequestTimeout is the timeout for realtime connection establishment
 	// and each subsequent operation.
 	RealtimeRequestTimeout time.Duration
@@ -209,13 +214,15 @@ type clientOptions struct {
 	// DisconnectedRetryTimeout is the time to wait after a disconnection before
 	// attempting an automatic reconnection, if still disconnected.
 	DisconnectedRetryTimeout time.Duration
+	SuspendedRetryTimeout    time.Duration
 	ChannelRetryTimeout      time.Duration
+	HTTPOpenTimeout          time.Duration
 
 	// Dial specifies the dial function for creating message connections used
 	// by Realtime.
 	//
 	// If Dial is nil, the default websocket connection is used.
-	Dial func(protocol string, u *url.URL) (proto.Conn, error)
+	Dial func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error)
 
 	// HTTPClient specifies the client used for HTTP communication by REST.
 	//
@@ -257,6 +264,12 @@ func (opts *clientOptions) realtimeRequestTimeout() time.Duration {
 	}
 	return defaultOptions.RealtimeRequestTimeout
 }
+func (opts *clientOptions) connectionStateTTL() time.Duration {
+	if opts.ConnectionStateTTL != 0 {
+		return opts.ConnectionStateTTL
+	}
+	return defaultOptions.ConnectionStateTTL
+}
 
 func (opts *clientOptions) disconnectedRetryTimeout() time.Duration {
 	if opts.DisconnectedRetryTimeout != 0 {
@@ -265,6 +278,20 @@ func (opts *clientOptions) disconnectedRetryTimeout() time.Duration {
 	return defaultOptions.DisconnectedRetryTimeout
 }
 
+func (opts *clientOptions) httpOpenTimeout() time.Duration {
+	if opts.HTTPOpenTimeout != 0 {
+		return opts.HTTPOpenTimeout
+	}
+	return defaultOptions.HTTPOpenTimeout
+}
+
+func (opts *clientOptions) suspendedRetryTimeout() time.Duration {
+	if opts.SuspendedRetryTimeout != 0 {
+		return opts.SuspendedRetryTimeout
+	}
+	return defaultOptions.SuspendedRetryTimeout
+}
+
 func (opts *clientOptions) restURL() string {
 	host := resolveHost(opts.RESTHost, opts.Environment, defaultOptions.RESTHost)
 	if opts.NoTLS {
@@ -624,6 +651,23 @@ func WithDisconnectedRetryTimeout(d time.Duration) ClientOption {
 	}
 }
 
+func WithHTTPOpenTimeout(d time.Duration) ClientOption {
+	return func(os *clientOptions) {
+		os.HTTPOpenTimeout = d
+	}
+}
+
+func WithSuspendedRetryTimeout(d time.Duration) ClientOption {
+	return func(os *clientOptions) {
+		os.SuspendedRetryTimeout = d
+	}
+}
+func WithConnectionStateTTL(d time.Duration) ClientOption {
+	return func(os *clientOptions) {
+		os.ConnectionStateTTL = d
+	}
+}
+
 func WithChannelRetryTimeout(d time.Duration) ClientOption {
 	return func(os *clientOptions) {
 		os.ChannelRetryTimeout = d
@@ -648,7 +692,7 @@ func WithHTTPClient(client *http.Client) ClientOption {
 	}
 }
 
-func WithDial(dial func(protocol string, u *url.URL) (proto.Conn, error)) ClientOption {
+func WithDial(dial func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error)) ClientOption {
 	return func(os *clientOptions) {
 		os.Dial = dial
 	}
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index 7a76069..08b0de3 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -28,7 +28,7 @@ func NewRealtime(options ...ClientOption) (*Realtime, error) {
 	c.rest = rest
 	c.Auth = rest.Auth
 	c.Channels = newChannels(c)
-	conn, err := newConn(c.opts(), rest.Auth, connCallbacks{
+	conn := newConn(c.opts(), rest.Auth, connCallbacks{
 		c.onChannelMsg,
 		c.onReconnected,
 		c.onReconnectionFailed,
@@ -36,9 +36,6 @@ func NewRealtime(options ...ClientOption) (*Realtime, error) {
 	conn.internalEmitter.OnAll(func(change ConnectionStateChange) {
 		c.Channels.broadcastConnStateChange(change)
 	})
-	if err != nil {
-		return nil, err
-	}
 	c.Connection = conn
 	return c, nil
 }
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index f0f5995..e53e2e0 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -34,6 +34,12 @@ const (
 type Connection struct {
 	mtx sync.Mutex
 
+	// on setConn we write to conn with mtx protection, however in eventLoop we
+	// read conn unprotected, this is racy because now we establish connection in a
+	// separate goroutine.
+	//
+	// using mtx to protect reads in eventLoop causes a deadlock.
+	connMtx sync.Mutex
 	ConnectionEventEmitter
 
 	state           ConnectionState
@@ -59,6 +65,7 @@ type Connection struct {
 	// reauthorizing tracks if the current reconnection attempt is happening
 	// after a reauthorization, to avoid re-reauthorizing.
 	reauthorizing bool
+	arg           connArgs
 }
 
 type connCallbacks struct {
@@ -73,7 +80,7 @@ type connCallbacks struct {
 	onReconnectionFailed func(*proto.ErrorInfo)
 }
 
-func newConn(opts *clientOptions, auth *Auth, callbacks connCallbacks) (*Connection, error) {
+func newConn(opts *clientOptions, auth *Auth, callbacks connCallbacks) *Connection {
 	c := &Connection{
 
 		ConnectionEventEmitter: ConnectionEventEmitter{newEventEmitter(auth.logger())},
@@ -87,34 +94,80 @@ func newConn(opts *clientOptions, auth *Auth, callbacks connCallbacks) (*Connect
 	}
 	c.queue = newMsgQueue(c)
 	if !opts.NoConnect {
-		if _, err := c.connect(false); err != nil {
-			return nil, err
-		}
+		go func() {
+			lg := opts.Logger.sugar()
+			lg.Info("Trying to establish a connection asynchronously")
+			if _, err := c.connect(connArgs{}); err != nil {
+				lg.Errorf("Failed to open connection with err:%v", err)
+			}
+		}()
 	}
-	return c, nil
+	return c
 }
 
-func (c *Connection) dial(proto string, u *url.URL) (proto.Conn, error) {
+func (c *Connection) dial(proto string, u *url.URL) (conn proto.Conn, err error) {
+	lg := c.logger().sugar()
+	start := time.Now()
+	lg.Debugf("Dial protocol=%q url %q ", proto, u.String())
 	// (RTN23b)
 	query := u.Query()
 	query.Add("heartbeats", "true")
 	u.RawQuery = query.Encode()
+	timeout := c.opts.realtimeRequestTimeout()
 	if c.opts.Dial != nil {
-		return c.opts.Dial(proto, u)
+		conn, err = c.opts.Dial(proto, u, timeout)
+	} else {
+		conn, err = ablyutil.DialWebsocket(proto, u, timeout)
+	}
+	if err != nil {
+		lg.Debugf("Dial Failed in %v with %v", time.Since(start), err)
+		return nil, err
+	}
+	lg.Debugf("Dial success in %v", time.Since(start))
+	return conn, err
+}
+
+func (c *Connection) connectAfterSuspension(arg connArgs) (Result, error) {
+	retryIn := c.opts.suspendedRetryTimeout()
+	lg := c.logger().sugar()
+	lg.Debugf("Attemting to periodically establish connection after suspension with timeout %v", retryIn)
+	tick := time.NewTicker(retryIn)
+	for {
+		<-tick.C
+		res, err := c.connectWith(arg)
+		if err != nil {
+			if recoverable(err) {
+				c.setState(ConnectionStateSuspended, err, retryIn)
+				continue
+			}
+			return nil, c.setState(ConnectionStateFailed, err, 0)
+		}
+		return res, nil
 	}
-	return ablyutil.DialWebsocket(proto, u)
+}
+
+// recoverable returns true if err is recoverable, err is from making a
+// connection
+func recoverable(err error) bool {
+	var e *ErrorInfo
+	if errors.As(err, &e) {
+		return !(40000 <= e.Code && e.Code < 50000)
+	}
+	return true
 }
 
 // Connect attempts to move the connection to the CONNECTED state, if it
 // can and if it isn't already.
 func (c *Connection) Connect() {
-	c.mtx.Lock()
-	isActive := c.isActive()
-	c.mtx.Unlock()
-	if isActive {
-		return
-	}
-	c.connect(false)
+	go func() {
+		c.mtx.Lock()
+		isActive := c.isActive()
+		c.mtx.Unlock()
+		if isActive {
+			return
+		}
+		c.connect(connArgs{})
+	}()
 }
 
 // Close attempts to move the connection to the CLOSED state, if it can and if
@@ -123,18 +176,27 @@ func (c *Connection) Close() {
 	c.close()
 }
 
-func (c *Connection) connect(result bool) (Result, error) {
+func (c *Connection) connect(arg connArgs) (Result, error) {
 	c.mtx.Lock()
-	mode := c.getMode()
+	arg.mode = c.getMode()
 	c.mtx.Unlock()
-	return c.connectWith(result, mode)
+	return c.connectWithRetryLoop(arg)
+}
+
+type connArgs struct {
+	lastActivityAt time.Time
+	connDetails    *proto.ConnectionDetails
+	result         bool
+	dialOnce       bool
+	mode           connectionMode
+	retryIn        time.Duration
 }
 
-func (c *Connection) reconnect(lastActivityAt time.Time, connDetails *proto.ConnectionDetails, result bool) (Result, error) {
+func (c *Connection) reconnect(arg connArgs) (Result, error) {
 	c.mtx.Lock()
 
 	var mode connectionMode
-	if connDetails != nil && c.opts.Now().Sub(lastActivityAt) >= time.Duration(connDetails.ConnectionStateTTL+connDetails.MaxIdleInterval) {
+	if arg.connDetails != nil && c.opts.Now().Sub(arg.lastActivityAt) >= time.Duration(arg.connDetails.ConnectionStateTTL+arg.connDetails.MaxIdleInterval) {
 		// RTN15g
 		c.msgSerial = 0
 		c.key = ""
@@ -146,8 +208,8 @@ func (c *Connection) reconnect(lastActivityAt time.Time, connDetails *proto.Conn
 	}
 
 	c.mtx.Unlock()
-
-	r, err := c.connectWith(result, mode)
+	arg.mode = mode
+	r, err := c.connectWithRetryLoop(arg)
 	if err != nil {
 		return nil, err
 	}
@@ -208,59 +270,119 @@ func (c *Connection) params(mode connectionMode) (url.Values, error) {
 	return query, nil
 }
 
-func (c *Connection) connectWith(result bool, mode connectionMode) (Result, error) {
+const connectionStateTTLErrFmt = "Exceeded connectionStateTtl=%v while in DISCONNECTED state"
+
+func (c *Connection) connectWithRetryLoop(arg connArgs) (Result, error) {
+	lg := c.logger().sugar()
+	res, err := c.connectWith(arg)
+	if err == nil {
+		return res, nil
+	}
+	if arg.dialOnce || !recoverable(err) {
+		return nil, c.setState(ConnectionStateFailed, err, 0)
+	}
+
+	lg.Errorf("Received recoverable error %v", err)
+	retryIn := c.opts.disconnectedRetryTimeout()
+	c.setState(ConnectionStateDisconnected, err, retryIn)
+
+	// The initial DISCONNECTED event has been fired. If we reach stateTTL without
+	// any state changes, we transition to SUSPENDED state
+	stateTTL := c.opts.connectionStateTTL()
+	stateDeadline := time.NewTimer(stateTTL)
+	defer stateDeadline.Stop()
+
+	next := time.NewTimer(retryIn)
+	reset := func() {
+		lg.Debugf("Retry in %v", retryIn)
+		next.Reset(retryIn)
+	}
+	defer next.Stop()
+	for {
+		select {
+		case <-stateDeadline.C:
+			// (RTN14e)
+			lg.Debug("Transition to SUSPENDED state")
+			err := fmt.Errorf(connectionStateTTLErrFmt, stateTTL)
+			c.setState(ConnectionStateSuspended, err, c.opts.suspendedRetryTimeout())
+			// (RTN14f)
+			lg.Debug("Reached SUSPENDED state while opening connection")
+			return c.connectAfterSuspension(arg)
+		case <-next.C:
+			lg.Debug("Attemting to open connection")
+			res, err := c.connectWith(arg)
+			if err == nil {
+				return res, nil
+			}
+			if recoverable(err) {
+				lg.Errorf("Received recoverable error %v", err)
+				// No need to reset stateDeadline as we are still in DISCONNECTED state. so
+				// another DISCONNECTED implies we haven't transitioned from DISCONNECTED
+				// state
+				c.setState(ConnectionStateDisconnected, err, retryIn)
+				reset()
+				continue
+			}
+			return nil, c.setState(ConnectionStateFailed, err, 0)
+		}
+	}
+}
+
+func (c *Connection) connectWith(arg connArgs) (Result, error) {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 	if !c.isActive() {
-		c.lockSetState(ConnectionStateConnecting, nil)
+		c.lockSetState(ConnectionStateConnecting, nil, 0)
 	}
 	u, err := url.Parse(c.opts.realtimeURL())
 	if err != nil {
-		return nil, c.lockSetState(ConnectionStateFailed, err)
+		return nil, err
 	}
 	var res Result
-	if result {
+	if arg.result {
 		res = c.internalEmitter.listenResult(
 			ConnectionStateConnected, // expected state
 			ConnectionStateFailed,
 			ConnectionStateDisconnected,
 		)
 	}
-	query, err := c.params(mode)
+	query, err := c.params(arg.mode)
 	if err != nil {
-		return nil, c.lockSetState(ConnectionStateFailed, err)
+		return nil, err
 	}
 	u.RawQuery = query.Encode()
 	proto := c.opts.protocol()
 	conn, err := c.dial(proto, u)
 	if err != nil {
-		return nil, c.lockSetState(ConnectionStateFailed, err)
+		return nil, err
 	}
 	if c.logger().Is(LogVerbose) {
 		c.setConn(verboseConn{conn: conn, logger: c.logger()})
 	} else {
 		c.setConn(conn)
 	}
-	c.reconnecting = mode == recoveryMode || mode == resumeMode
+	c.reconnecting = arg.mode == recoveryMode || arg.mode == resumeMode
+	c.arg = arg
 	return res, nil
 }
 
+func (c *Connection) connectionStateTTL() time.Duration {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	if c.arg.connDetails != nil && c.arg.connDetails.ConnectionStateTTL != 0 {
+		return time.Duration(c.arg.connDetails.ConnectionStateTTL)
+	}
+	return c.opts.connectionStateTTL()
+}
+
 func (c *Connection) close() {
 	c.mtx.Lock()
 	c.key, c.id = "", "" //(RTN16c)
 	defer c.mtx.Unlock()
-	switch c.state {
-	case
-		ConnectionStateClosing,
-		ConnectionStateClosed,
-		ConnectionStateInitialized,
-		ConnectionStateFailed,
-		ConnectionStateDisconnected:
-
+	if c.state != ConnectionStateConnected {
 		return
 	}
-
-	c.lockSetState(ConnectionStateClosing, nil)
+	c.lockSetState(ConnectionStateClosing, nil, 0)
 	msg := &proto.ProtocolMessage{Action: proto.ActionClose}
 	c.updateSerial(msg, nil)
 
@@ -478,7 +600,9 @@ func (c *Connection) lockIsActive() bool {
 }
 
 func (c *Connection) setConn(conn proto.Conn) {
+	c.connMtx.Lock()
 	c.conn = conn
+	c.connMtx.Unlock()
 	go c.eventloop()
 }
 
@@ -500,7 +624,9 @@ func (c *Connection) eventloop() {
 			receiveTimeout := c.opts.realtimeRequestTimeout() + maxIdleInterval // RTN23a
 			deadline = c.opts.Now().Add(receiveTimeout)                         // RTNf23a
 		}
+		c.connMtx.Lock()
 		msg, err := c.conn.Receive(deadline)
+		c.connMtx.Unlock()
 		if err != nil {
 			c.mtx.Lock()
 			if c.state == ConnectionStateClosed {
@@ -509,9 +635,13 @@ func (c *Connection) eventloop() {
 			}
 
 			// RTN23a
-			c.lockSetState(ConnectionStateDisconnected, err)
+			c.lockSetState(ConnectionStateDisconnected, err, 0)
 			c.mtx.Unlock()
-			c.reconnect(lastActivityAt, connDetails, false)
+			arg := connArgs{
+				lastActivityAt: lastActivityAt,
+				connDetails:    connDetails,
+			}
+			c.reconnect(arg)
 			return
 		}
 		lastActivityAt = c.opts.Now()
@@ -532,6 +662,7 @@ func (c *Connection) eventloop() {
 			c.pending.Nack(msg.MsgSerial, msg.Count, newErrorFromProto(msg.Error))
 			c.mtx.Unlock()
 		case proto.ActionError:
+
 			if msg.Channel != "" {
 				c.callbacks.onChannelMsg(msg)
 				break
@@ -545,9 +676,15 @@ func (c *Connection) eventloop() {
 					c.lockedReauthorizationFailed(newErrorFromProto(msg.Error))
 					c.mtx.Unlock()
 					return
-				} else {
-					// TODO: RTN14b; may reuse c.reauthorize from RTN15h2.
 				}
+				// RTN14b
+				c.mtx.Unlock()
+				c.reauthorize(connArgs{
+					lastActivityAt: lastActivityAt,
+					connDetails:    connDetails,
+					dialOnce:       true,
+				})
+				return
 			}
 			c.mtx.Unlock()
 
@@ -573,7 +710,7 @@ func (c *Connection) eventloop() {
 			previousID := c.id
 			c.id = msg.ConnectionID
 			c.msgSerial = 0
-			if reconnecting && mode == recoveryMode {
+			if reconnecting && mode == recoveryMode && msg.Error == nil {
 				// we are setting msgSerial as per (RTN16f)
 				msgSerial, err := strconv.ParseInt(strings.Split(c.opts.Recover, ":")[2], 10, 64)
 				if err != nil {
@@ -582,11 +719,12 @@ func (c *Connection) eventloop() {
 				c.msgSerial = msgSerial
 			}
 			c.setSerial(-1)
+
 			c.mtx.Unlock()
 			if reconnecting {
 				// (RTN15c1) (RTN15c2)
 				c.mtx.Lock()
-				c.lockSetState(ConnectionStateConnected, newErrorFromProto(msg.Error))
+				c.lockSetState(ConnectionStateConnected, newErrorFromProto(msg.Error), 0)
 				c.mtx.Unlock()
 				if previousID != msg.ConnectionID {
 					// (RTN15c3)
@@ -598,7 +736,7 @@ func (c *Connection) eventloop() {
 			} else {
 				// preserve old behavior.
 				c.mtx.Lock()
-				c.lockSetState(ConnectionStateConnected, nil)
+				c.lockSetState(ConnectionStateConnected, nil, 0)
 				c.mtx.Unlock()
 			}
 			c.queue.Flush()
@@ -617,12 +755,15 @@ func (c *Connection) eventloop() {
 			}
 
 			// RTN15h2
-			c.reauthorize(lastActivityAt, connDetails)
+			c.reauthorize(connArgs{
+				lastActivityAt: lastActivityAt,
+				connDetails:    connDetails,
+			})
 			return
 		case proto.ActionClosed:
 			c.mtx.Lock()
 			c.id, c.key = "", "" //(RTN16c)
-			c.lockSetState(ConnectionStateClosed, nil)
+			c.lockSetState(ConnectionStateClosed, nil, 0)
 			c.mtx.Unlock()
 			if c.conn != nil {
 				c.conn.Close()
@@ -640,7 +781,7 @@ func (c *Connection) failedConnSideEffects(err *proto.ErrorInfo) {
 		c.reauthorizing = false
 		c.callbacks.onReconnectionFailed(err)
 	}
-	c.lockSetState(ConnectionStateFailed, newErrorFromProto(err))
+	c.lockSetState(ConnectionStateFailed, newErrorFromProto(err), 0)
 	c.mtx.Unlock()
 	c.queue.Fail(newErrorFromProto(err))
 	if c.conn != nil {
@@ -648,9 +789,10 @@ func (c *Connection) failedConnSideEffects(err *proto.ErrorInfo) {
 	}
 }
 
-func (c *Connection) reauthorize(lastActivityAt time.Time, connDetails *proto.ConnectionDetails) {
+func (c *Connection) reauthorize(arg connArgs) {
 	c.mtx.Lock()
 	_, err := c.auth.reauthorize()
+
 	if err != nil {
 		c.lockedReauthorizationFailed(err)
 		c.mtx.Unlock()
@@ -661,12 +803,11 @@ func (c *Connection) reauthorize(lastActivityAt time.Time, connDetails *proto.Co
 	// reconnecting will use the new token.
 	c.reauthorizing = true
 	c.mtx.Unlock()
-	c.reconnect(lastActivityAt, connDetails, false)
+	c.reconnect(arg)
 }
 
 func (c *Connection) lockedReauthorizationFailed(err error) {
-	c.lockSetState(ConnectionStateDisconnected, err)
-	// TODO: RTN14d
+	c.lockSetState(ConnectionStateDisconnected, err, 0)
 }
 
 type verboseConn struct {
@@ -693,13 +834,13 @@ func (vc verboseConn) Close() error {
 	return vc.conn.Close()
 }
 
-func (c *Connection) setState(state ConnectionState, err error) error {
+func (c *Connection) setState(state ConnectionState, err error, retryIn time.Duration) error {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
-	return c.lockSetState(state, err)
+	return c.lockSetState(state, err, retryIn)
 }
 
-func (c *Connection) lockSetState(state ConnectionState, err error) error {
+func (c *Connection) lockSetState(state ConnectionState, err error, retryIn time.Duration) error {
 	previous := c.state
 	changed := c.state != state
 	c.state = state
@@ -708,6 +849,7 @@ func (c *Connection) lockSetState(state ConnectionState, err error) error {
 		Current:  c.state,
 		Previous: previous,
 		Reason:   c.errorReason,
+		RetryIn:  retryIn,
 	}
 	if !changed {
 		change.Event = ConnectionEventUpdated
diff --git a/ably/state.go b/ably/state.go
index 5086092..5bbc4bd 100644
--- a/ably/state.go
+++ b/ably/state.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"sort"
 	"sync"
+	"time"
 
 	"github.com/ably/ably-go/ably/proto"
 )
@@ -380,6 +381,7 @@ type ConnectionStateChange struct {
 	Current  ConnectionState
 	Event    ConnectionEvent
 	Previous ConnectionState
+	RetryIn  time.Duration //RTN14d, TA2
 	// Reason, if any, is an error that caused the state change.
 	Reason *ErrorInfo
 }


Test patch:
diff --git a/ably/ablytest/recorders.go b/ably/ablytest/recorders.go
index c773496..87feb50 100644
--- a/ably/ablytest/recorders.go
+++ b/ably/ablytest/recorders.go
@@ -209,8 +209,8 @@ func MessagePipeWithNowFunc(now func() time.Time) MessagePipeOption {
 	}
 }
 
-func MessagePipe(in <-chan *proto.ProtocolMessage, out chan<- *proto.ProtocolMessage, opts ...MessagePipeOption) func(string, *url.URL) (proto.Conn, error) {
-	return func(proto string, u *url.URL) (proto.Conn, error) {
+func MessagePipe(in <-chan *proto.ProtocolMessage, out chan<- *proto.ProtocolMessage, opts ...MessagePipeOption) func(string, *url.URL, time.Duration) (proto.Conn, error) {
+	return func(proto string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 		pc := pipeConn{
 			in:  in,
 			out: out,
@@ -278,11 +278,11 @@ func NewMessageRecorder() *MessageRecorder {
 }
 
 // Dial
-func (rec *MessageRecorder) Dial(proto string, u *url.URL) (proto.Conn, error) {
+func (rec *MessageRecorder) Dial(proto string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 	rec.mu.Lock()
 	rec.url = append(rec.url, u)
 	rec.mu.Unlock()
-	conn, err := ablyutil.DialWebsocket(proto, u)
+	conn, err := ablyutil.DialWebsocket(proto, u, timeout)
 	if err != nil {
 		return nil, err
 	}
@@ -354,7 +354,7 @@ type HostRecorder struct {
 
 	mu         sync.Mutex
 	httpClient *http.Client
-	dialWS     func(string, *url.URL) (proto.Conn, error)
+	dialWS     func(string, *url.URL, time.Duration) (proto.Conn, error)
 }
 
 func NewRecorder(httpClient *http.Client) *HostRecorder {
@@ -368,9 +368,9 @@ func NewRecorder(httpClient *http.Client) *HostRecorder {
 		hr.addHost(addr)
 		return dial(network, addr)
 	}
-	hr.dialWS = func(proto string, u *url.URL) (proto.Conn, error) {
+	hr.dialWS = func(proto string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 		hr.addHost(u.Host)
-		return ablyutil.DialWebsocket(proto, u)
+		return ablyutil.DialWebsocket(proto, u, timeout)
 	}
 	return hr
 }
@@ -406,15 +406,15 @@ func body(p []byte) io.ReadCloser {
 // clientOptions.
 func DialFakeDisconnect(dial DialFunc) (_ DialFunc, disconnect func() error) {
 	if dial == nil {
-		dial = func(proto string, url *url.URL) (proto.Conn, error) {
-			return ablyutil.DialWebsocket(proto, url)
+		dial = func(proto string, url *url.URL, timeout time.Duration) (proto.Conn, error) {
+			return ablyutil.DialWebsocket(proto, url, timeout)
 		}
 	}
 
 	disconnectReq := make(chan chan<- error, 1)
 
-	return func(proto string, url *url.URL) (proto.Conn, error) {
-			conn, err := dial(proto, url)
+	return func(proto string, url *url.URL, timeout time.Duration) (proto.Conn, error) {
+			conn, err := dial(proto, url, timeout)
 			if err != nil {
 				return nil, err
 			}
@@ -431,7 +431,7 @@ func DialFakeDisconnect(dial DialFunc) (_ DialFunc, disconnect func() error) {
 		}
 }
 
-type DialFunc func(proto string, url *url.URL) (proto.Conn, error)
+type DialFunc func(proto string, url *url.URL, timeout time.Duration) (proto.Conn, error)
 
 type connWithFakeDisconnect struct {
 	conn          proto.Conn
diff --git a/ably/ablytest/resultgroup.go b/ably/ablytest/resultgroup.go
index e52e57c..829855f 100644
--- a/ably/ablytest/resultgroup.go
+++ b/ably/ablytest/resultgroup.go
@@ -55,7 +55,7 @@ func (rg *ResultGroup) Add(res ably.Result, err error) {
 	rg.wg.Add(1)
 	go func() {
 		err := res.Wait()
-		if err != nil {
+		if err != nil && err != (*ably.ErrorInfo)(nil) {
 			select {
 			case rg.errch <- err:
 			default:
diff --git a/ably/auth_test.go b/ably/auth_test.go
index 1f4e774..a7c965e 100644
--- a/ably/auth_test.go
+++ b/ably/auth_test.go
@@ -576,6 +576,9 @@ func TestAuth_RequestToken_PublishClientID(t *testing.T) {
 		}
 		client := app.NewRealtime(opts...)
 		defer safeclose(t, ablytest.FullRealtimeCloser(client))
+		if err = ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
+			t.Fatalf("Connect(): want err == nil got err=%v", err)
+		}
 		if id := client.Auth.ClientID(); id != cas.clientID {
 			t.Errorf("%d: want ClientID to be %q; got %s", i, cas.clientID, id)
 			continue
@@ -680,9 +683,6 @@ func TestAuth_ClientID(t *testing.T) {
 	time.Sleep(time.Duration(params.TTL) * time.Millisecond)
 	tokenExpiredAt := time.Now()
 
-	failed := make(ably.ConnStateChanges, 1)
-	client.Connection.On(ably.ConnectionEventFailed, failed.Receive)
-
 	// Allow some extra time for the server to reject our token.
 	err = nil
 	tokenErrorDeadline := tokenExpiredAt.Add(5 * time.Second)
@@ -716,12 +716,6 @@ func TestAuth_ClientID(t *testing.T) {
 	if state := client.Connection.State(); state != ably.ConnectionStateFailed {
 		t.Fatalf("want state=%q; got %q", ably.ConnectionStateFailed, state)
 	}
-
-	var state ably.ConnectionStateChange
-	ablytest.Soon.Recv(t, &state, failed, t.Fatalf)
-	if state.Current != ably.ConnectionStateFailed {
-		t.Fatalf("want state=%q; got %q", ably.ConnectionStateFailed, state.Current)
-	}
 }
 
 func TestAuth_CreateTokenRequest(t *testing.T) {
diff --git a/ably/export_test.go b/ably/export_test.go
index 3385e10..028bbd5 100644
--- a/ably/export_test.go
+++ b/ably/export_test.go
@@ -119,6 +119,8 @@ type ConnStateChanges = connStateChanges
 
 type ChannelStateChanges = channelStateChanges
 
+const ConnectionStateTTLErrFmt = connectionStateTTLErrFmt
+
 func DefaultFallbackHosts() []string {
 	return defaultFallbackHosts()
 }
diff --git a/ably/realtime_channel_test.go b/ably/realtime_channel_test.go
index 638d234..f135e81 100644
--- a/ably/realtime_channel_test.go
+++ b/ably/realtime_channel_test.go
@@ -40,7 +40,9 @@ func TestRealtimeChannel_Publish(t *testing.T) {
 	t.Parallel()
 	app, client := ablytest.NewRealtime(nil...)
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
-
+	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
+		t.Fatal(err)
+	}
 	channel := client.Channels.Get("test")
 	if err := channel.Publish(context.Background(), "hello", "world"); err != nil {
 		t.Fatalf("Publish()=%v", err)
@@ -53,7 +55,12 @@ func TestRealtimeChannel_Subscribe(t *testing.T) {
 	defer safeclose(t, ablytest.FullRealtimeCloser(client1), app)
 	client2 := app.NewRealtime(ably.WithEchoMessages(false))
 	defer safeclose(t, ablytest.FullRealtimeCloser(client2))
-
+	if err := ablytest.ConnWaiter(client1, client1.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
+		t.Fatal(err)
+	}
+	if err := ablytest.ConnWaiter(client2, client2.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
+		t.Fatal(err)
+	}
 	channel1 := client1.Channels.Get("test")
 	channel2 := client2.Channels.Get("test")
 
@@ -103,7 +110,9 @@ func TestRealtimeChannel_Detach(t *testing.T) {
 	t.Parallel()
 	app, client := ablytest.NewRealtime()
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
-
+	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
+		t.Fatal(err)
+	}
 	channel := client.Channels.Get("test")
 	sub, unsub, err := ablytest.ReceiveMessages(channel, "")
 	if err != nil {
@@ -152,9 +161,9 @@ func TestRealtimeChannel_AttachWhileDisconnected(t *testing.T) {
 
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			<-allowDial
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF}, err
 		}))
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
diff --git a/ably/realtime_client_test.go b/ably/realtime_client_test.go
index 8d32ab8..06b47ae 100644
--- a/ably/realtime_client_test.go
+++ b/ably/realtime_client_test.go
@@ -3,67 +3,47 @@ package ably_test
 import (
 	"context"
 	"fmt"
+	"net"
+	"net/url"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/ably/ably-go/ably"
 	"github.com/ably/ably-go/ably/ablytest"
+	"github.com/ably/ably-go/ably/proto"
 )
 
 func TestRealtime_RealtimeHost(t *testing.T) {
 	t.Parallel()
-	httpClient := ablytest.NewHTTPClient()
-	app, err := ablytest.NewSandbox(nil)
-	if err != nil {
-		t.Fatalf("NewSandbox=%s", err)
-	}
-	defer safeclose(t, app)
-
-	rec := ablytest.NewRecorder(httpClient)
 	hosts := []string{
 		"127.0.0.1",
 		"localhost",
 		"::1",
 	}
-	var errorsRec ablytest.ConnErrorsRecorder
 	for _, host := range hosts {
-		opts := rec.Options(host)
-		client, err := ably.NewRealtime(app.Options(opts...)...)
+		dial := make(chan string, 1)
+		client, err := ably.NewRealtime(
+			ably.WithKey("xxx:xxx"),
+			ably.WithRealtimeHost(host),
+			ably.WithAutoConnect(false),
+			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+				dial <- u.Host
+				return ablytest.MessagePipe(nil, nil)(protocol, u, timeout)
+			}),
+		)
 		if err != nil {
-			t.Errorf("NewRealtime=%s (host=%s)", err, host)
-			continue
-		}
-		{
-			off := errorsRec.Listen(client)
-			defer off()
-		}
-		if state := client.Connection.State(); state != ably.ConnectionStateInitialized {
-			t.Errorf("want state=%v; got %s", ably.ConnectionStateInitialized, state)
-			continue
-		}
-		if err := checkError(80000, ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait()); err != nil {
-			t.Errorf("%s (host=%s)", err, host)
-			continue
+			t.Fatal(err)
 		}
-		if err := checkError(80000, ablytest.FullRealtimeCloser(client).Close()); err != nil {
-			t.Errorf("%s (host=%s)", err, host)
-			continue
+		client.Connect()
+		var recordedHost string
+		ablytest.Instantly.Recv(t, &recordedHost, dial, t.Fatalf)
+		h, _, err := net.SplitHostPort(recordedHost)
+		if err != nil {
+			t.Fatal(err)
 		}
-		if _, ok := rec.Hosts[host]; !ok {
-			t.Errorf("host %s was not recorded (recorded %v)", host, rec.Hosts)
-		}
-	}
-
-	var errors []*ably.ErrorInfo
-	if !ablytest.Soon.IsTrue(func() bool {
-		errors = errorsRec.Errors()
-		return len(errors) == len(hosts)
-	}) {
-		t.Fatalf("want len(errors)=%d; got %d", len(hosts), len(errors))
-	}
-	for i, err := range errors {
-		if err := checkError(80000, err); err != nil {
-			t.Errorf("%s (host=%s)", err, hosts[i])
+		if h != host {
+			t.Errorf(" expected %q got %q", host, h)
 		}
 	}
 }
@@ -107,6 +87,10 @@ func TestRealtime_multiple(t *testing.T) {
 				all.Add(nil, err)
 				return
 			}
+			if err = ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
+				t.Error(err)
+				return
+			}
 			var rg ablytest.ResultGroup
 			rg.Add(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil)
 			for j := 0; j < 10; j++ {
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index a2c35cd..970c5a7 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -1,14 +1,21 @@
 package ably_test
 
 import (
+	"bytes"
 	"context"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
+	"log"
+	"net"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
+	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -183,8 +190,8 @@ func TestRealtimeConn_RTN15a_ReconnectOnEOF(t *testing.T) {
 
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
-			c, err := ablyutil.DialWebsocket(protocol, u)
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF}, err
 		}))
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
@@ -311,7 +318,7 @@ func TestRealtimeConn_RTN15b(t *testing.T) {
 	gotDial := make(chan chan struct{})
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			m := &meta{dial: u}
 			metaList = append(metaList, m)
 			if len(metaList) > 1 {
@@ -319,7 +326,7 @@ func TestRealtimeConn_RTN15b(t *testing.T) {
 				gotDial <- goOn
 				<-goOn
 			}
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
 				m.messages = append(m.messages, msg)
 			}}, err
@@ -441,7 +448,7 @@ func TestRealtimeConn_RTN15c1(t *testing.T) {
 
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			m := &meta{dial: u}
 			metaList = append(metaList, m)
 			if len(metaList) > 1 {
@@ -449,7 +456,7 @@ func TestRealtimeConn_RTN15c1(t *testing.T) {
 				gotDial <- goOn
 				<-goOn
 			}
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
 				m.messages = append(m.messages, msg)
 			}}, err
@@ -557,7 +564,7 @@ func TestRealtimeConn_RTN15c2(t *testing.T) {
 	gotDial := make(chan chan struct{})
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			m := &meta{dial: u}
 			metaList = append(metaList, m)
 			if len(metaList) > 1 {
@@ -565,7 +572,7 @@ func TestRealtimeConn_RTN15c2(t *testing.T) {
 				gotDial <- goOn
 				<-goOn
 			}
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
 				if len(metaList) == 2 && len(m.messages) == 0 {
 					msg.Error = errInfo
@@ -685,7 +692,7 @@ func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 	gotDial := make(chan chan struct{})
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			m := &meta{dial: u}
 			metaList = append(metaList, m)
 			if len(metaList) > 1 {
@@ -693,7 +700,7 @@ func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 				gotDial <- goOn
 				<-goOn
 			}
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
 				if len(metaList) == 2 && len(m.messages) == 0 {
 					msg.Error = errInfo
@@ -793,7 +800,7 @@ func TestRealtimeConn_RTN15c3_attaching(t *testing.T) {
 	gotDial := make(chan chan struct{})
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			m := &meta{dial: u}
 			metaList = append(metaList, m)
 			if len(metaList) > 1 {
@@ -801,7 +808,7 @@ func TestRealtimeConn_RTN15c3_attaching(t *testing.T) {
 				gotDial <- goOn
 				<-goOn
 			}
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
 				if len(metaList) == 2 && len(m.messages) == 0 {
 					msg.Error = errInfo
@@ -902,7 +909,7 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 	gotDial := make(chan chan struct{})
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			m := &meta{dial: u}
 			metaList = append(metaList, m)
 			if len(metaList) > 1 {
@@ -910,7 +917,7 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 				gotDial <- goOn
 				<-goOn
 			}
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
 				if len(metaList) == 2 && len(m.messages) == 0 {
 					msg.Action = proto.ActionError
@@ -1002,9 +1009,9 @@ func TestRealtimeConn_RTN15d_MessageRecovery(t *testing.T) {
 
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			<-allowDial
-			c, err := ablyutil.DialWebsocket(protocol, u)
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF}, err
 		}))
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
@@ -1074,8 +1081,8 @@ func TestRealtimeConn_RTN15e_ConnKeyUpdatedOnReconnect(t *testing.T) {
 
 	app, client := ablytest.NewRealtime(
 		ably.WithAutoConnect(false),
-		ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
-			c, err := ablyutil.DialWebsocket(protocol, u)
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+			c, err := ablyutil.DialWebsocket(protocol, u, timeout)
 			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF}, err
 		}))
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
@@ -1144,7 +1151,7 @@ func TestRealtimeConn_RTN15g_NewConnectionOnStateLost(t *testing.T) {
 		ably.WithAutoConnect(false),
 		ably.WithToken("fake:token"),
 		ably.WithNow(now),
-		ably.WithDial(func(p string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(p string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			in = make(chan *proto.ProtocolMessage, 1)
 			in <- &proto.ProtocolMessage{
 				Action:            proto.ActionConnected,
@@ -1153,7 +1160,7 @@ func TestRealtimeConn_RTN15g_NewConnectionOnStateLost(t *testing.T) {
 			}
 			breakConn = func() { close(in) }
 			dials <- u
-			return ablytest.MessagePipe(in, out)(p, u)
+			return ablytest.MessagePipe(in, out)(p, u, timeout)
 		}))
 
 	connIDs <- "conn-1"
@@ -1372,9 +1379,9 @@ func TestRealtimeConn_RTN15h2_ReauthWithBadToken(t *testing.T) {
 		ably.WithAuthCallback(func(context.Context, ably.TokenParams) (ably.Tokener, error) {
 			return ably.TokenString("bad:token"), nil
 		}),
-		ably.WithDial(func(proto string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(proto string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			dials <- u
-			return ablytest.MessagePipe(in, out)(proto, u)
+			return ablytest.MessagePipe(in, out)(proto, u, timeout)
 		}))
 
 	in <- &proto.ProtocolMessage{
@@ -1459,9 +1466,9 @@ func TestRealtimeConn_RTN15h2_Success(t *testing.T) {
 		ably.WithAuthCallback(func(context.Context, ably.TokenParams) (ably.Tokener, error) {
 			return ably.TokenString("good:token"), nil
 		}),
-		ably.WithDial(func(proto string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(proto string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			dials <- u
-			return ablytest.MessagePipe(in, out)(proto, u)
+			return ablytest.MessagePipe(in, out)(proto, u, timeout)
 		}))
 
 	in <- &proto.ProtocolMessage{
@@ -1638,9 +1645,6 @@ func TestRealtimeConn_RTN16(t *testing.T) {
 		if expected, got := prevMsgSerial, client.Connection.MsgSerial(); expected != got {
 			t.Errorf("expected %d got %d", expected, got)
 		}
-		if true {
-			return
-		}
 	}
 
 	{ //(RTN16c)
@@ -1663,9 +1667,9 @@ func TestRealtimeConn_RTN16(t *testing.T) {
 		fakeRecoveryKey := "_____!ablygo_test_fake-key____:5:3"
 		client2 := app.NewRealtime(
 			ably.WithRecover(fakeRecoveryKey),
-			ably.WithDial(func(protocol string, u *url.URL) (proto.Conn, error) {
+			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 				query = u.Query()
-				return ablyutil.DialWebsocket(protocol, u)
+				return ablyutil.DialWebsocket(protocol, u, timeout)
 			}))
 		defer safeclose(t, ablytest.FullRealtimeCloser(client2))
 		err = ablytest.ConnWaiter(client2, client2.Connect, ably.ConnectionEventConnected).Wait()
@@ -1739,7 +1743,7 @@ func TestRealtimeConn_RTN23(t *testing.T) {
 		ably.WithAutoConnect(false),
 		ably.WithToken("fake:token"),
 		ably.WithRealtimeRequestTimeout(realtimeRequestTimeout),
-		ably.WithDial(func(p string, u *url.URL) (proto.Conn, error) {
+		ably.WithDial(func(p string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
 			in = make(chan *proto.ProtocolMessage, 1)
 			in <- &proto.ProtocolMessage{
 				Action:            proto.ActionConnected,
@@ -1749,8 +1753,12 @@ func TestRealtimeConn_RTN23(t *testing.T) {
 			dials <- u
 			return ablytest.MessagePipe(in, nil,
 				ablytest.MessagePipeWithNowFunc(time.Now),
-			)(p, u)
+			)(p, u, timeout)
 		}))
+	disconnected := make(chan *ably.ErrorInfo, 1)
+	c.Connection.Once(ably.ConnectionEventDisconnected, func(e ably.ConnectionStateChange) {
+		disconnected <- e.Reason
+	})
 	err := ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait()
 	if err != nil {
 		t.Fatal(err)
@@ -1766,10 +1774,6 @@ func TestRealtimeConn_RTN23(t *testing.T) {
 	maxIdleInterval := time.Duration(connDetails.MaxIdleInterval)
 	receiveTimeout := realtimeRequestTimeout + maxIdleInterval
 
-	disconnected := make(chan *ably.ErrorInfo, 1)
-	c.Connection.Once(ably.ConnectionEventDisconnected, func(e ably.ConnectionStateChange) {
-		disconnected <- e.Reason
-	})
 	in <- &proto.ProtocolMessage{
 		Action: proto.ActionHeartbeat,
 	}
@@ -1786,3 +1790,307 @@ func TestRealtimeConn_RTN23(t *testing.T) {
 		t.Errorf("expected %q to contain timeout", reason.Error())
 	}
 }
+
+type writerLogger struct {
+	w io.Writer
+}
+
+func (w *writerLogger) Print(level ably.LogLevel, v ...interface{}) {
+	fmt.Fprint(w.w, v...)
+}
+
+func (w *writerLogger) Printf(level ably.LogLevel, format string, v ...interface{}) {
+	fmt.Fprintf(w.w, format, v...)
+}
+
+func TestRealtimeConn_RTN14c(t *testing.T) {
+	t.Parallel()
+	ts := httptest.NewUnstartedServer(nil)
+	reqTimeout := 5 * time.Millisecond
+	ts.Config = &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			time.Sleep(reqTimeout + 5*time.Millisecond)
+		}),
+	}
+	ts.TLS = &tls.Config{
+		GetConfigForClient: func(chi *tls.ClientHelloInfo) (*tls.Config, error) {
+			time.Sleep(reqTimeout * 2)
+			return nil, context.DeadlineExceeded
+		},
+	}
+	ts.StartTLS()
+	defer ts.Close()
+
+	host, port, err := net.SplitHostPort(strings.TrimPrefix(ts.URL, "https://"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	p, _ := strconv.Atoi(port)
+	var buf bytes.Buffer
+	lg := &writerLogger{w: &buf}
+	client, err := ably.NewRealtime(
+		ably.WithKey("xxx:xxx"),
+		ably.WithRealtimeHost(host),
+		ably.WithTLSPort(p),
+		ably.WithRealtimeRequestTimeout(reqTimeout),
+		ably.WithLogHandler(lg),
+		ably.WithLogLevel(ably.LogDebug),
+		ably.WithAutoConnect(false),
+	)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer client.Close()
+	ok := make(chan struct{}, 1)
+	client.Connection.On(ably.ConnectionEventDisconnected, func(csc ably.ConnectionStateChange) {
+		ok <- struct{}{}
+	})
+	client.Connect()
+	ablytest.Soon.Recv(t, nil, ok, t.Fatalf)
+	sub := "Dial Failed in "
+	x := buf.String()
+	idx := strings.Index(x, sub)
+	if idx == -1 {
+		t.Fatal("Missing Dial i/o timeout error")
+	}
+	x = x[idx+len(sub):]
+	x = strings.TrimSpace(x)
+	x = strings.Split(x, " ")[0]
+
+	d, err := time.ParseDuration(x)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// We can't be precise here so just an estimate. Manual tests ranges between 5ms -10ms
+	// TODO: Find a proper way to record Dial i/o timeouts duration.
+	if d < reqTimeout || d > reqTimeout+2 {
+		t.Errorf("expected i/o timeout to be %v got %v", reqTimeout, d)
+	}
+}
+
+func TestRealtimeConn_RTN14a(t *testing.T) {
+	t.Skip(`
+	Currently its impossible to implement/test this.
+	See https://github.com/ably/docs/issues/984 for details
+	`)
+}
+
+func TestRealtimeConn_RTN14b(t *testing.T) {
+	t.Parallel()
+	t.Run("renewable token that fails to renew with token error", func(t *testing.T) {
+		t.Parallel()
+		in := make(chan *proto.ProtocolMessage, 1)
+		out := make(chan *proto.ProtocolMessage, 16)
+		var reauth atomic.Value
+		reauth.Store(int(0))
+		bad := "bad token request"
+		c, _ := ably.NewRealtime(
+			ably.WithAutoConnect(false),
+			ably.WithAuthCallback(func(context.Context, ably.TokenParams) (ably.Tokener, error) {
+				reauth.Store(reauth.Load().(int) + 1)
+				if reauth.Load().(int) > 1 {
+					return nil, errors.New(bad)
+				}
+				return ably.TokenString("fake:token"), nil
+			}),
+			ably.WithDial(ablytest.MessagePipe(in, out)))
+		// Get the connection to CONNECTED.
+		tokenError := &proto.ErrorInfo{
+			StatusCode: http.StatusUnauthorized,
+			Code:       40140,
+		}
+		in <- &proto.ProtocolMessage{
+			Action:            proto.ActionError,
+			ConnectionDetails: &proto.ConnectionDetails{},
+			Error:             tokenError,
+		}
+		change := make(ably.ConnStateChanges, 1)
+		c.Connection.OnAll(change.Receive)
+		c.Connect()
+		var state ably.ConnectionStateChange
+		ablytest.Instantly.Recv(t, nil, change, t.Fatalf) // Skip CONNECTING
+		ablytest.Instantly.Recv(t, &state, change, t.Fatalf)
+		if expect, got := ably.ConnectionStateDisconnected, state.Current; expect != got {
+			t.Errorf("expected %v got %v", expect, got)
+		}
+		if expect, got := bad, c.Connection.ErrorReason(); !strings.Contains(got.Error(), expect) {
+			t.Errorf("expected %v to contain %q", got, expect)
+		}
+		n := reauth.Load().(int) - 1 // we remove the first attempt
+		if n != 1 {
+			t.Errorf("expected re authorization to happen once but happened %d", n)
+		}
+	})
+	t.Run("renewable token, consecutive token errors", func(t *testing.T) {
+		t.Parallel()
+		in := make(chan *proto.ProtocolMessage, 1)
+		out := make(chan *proto.ProtocolMessage, 16)
+		var reauth atomic.Value
+		reauth.Store(int(0))
+		var dials atomic.Value
+		dials.Store(int(0))
+		c, _ := ably.NewRealtime(
+			ably.WithAutoConnect(false),
+			ably.WithAuthCallback(func(context.Context, ably.TokenParams) (ably.Tokener, error) {
+				reauth.Store(reauth.Load().(int) + 1)
+				return ably.TokenString("fake:token"), nil
+			}),
+			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+				dials.Store(dials.Load().(int) + 1)
+				return ablytest.MessagePipe(in, out)(protocol, u, timeout)
+			}))
+		// Get the connection to CONNECTED.
+		tokenError := &proto.ErrorInfo{
+			StatusCode: http.StatusUnauthorized,
+			Code:       40140,
+		}
+		in <- &proto.ProtocolMessage{
+			Action:            proto.ActionError,
+			ConnectionDetails: &proto.ConnectionDetails{},
+			Error:             tokenError,
+		}
+		change := make(ably.ConnStateChanges, 1)
+		c.Connection.OnAll(change.Receive)
+		c.Connect()
+		bad := &proto.ErrorInfo{
+			StatusCode: http.StatusUnauthorized,
+			Code:       40140,
+			Message:    "bad token request",
+		}
+		in <- &proto.ProtocolMessage{
+			Action:            proto.ActionError,
+			ConnectionDetails: &proto.ConnectionDetails{},
+			Error:             bad,
+		}
+		var state ably.ConnectionStateChange
+		ablytest.Instantly.Recv(t, nil, change, t.Fatalf) // skip CONNECTING
+		ablytest.Instantly.Recv(t, &state, change, t.Fatalf)
+		if expect, got := ably.ConnectionStateDisconnected, state.Current; expect != got {
+			t.Errorf("expected %v got %v", expect, got)
+		}
+		if expect, got := bad.Message, c.Connection.ErrorReason(); !strings.Contains(got.Error(), expect) {
+			t.Errorf("expected %v to contain %q", got, expect)
+		}
+		n := reauth.Load().(int)
+		if n != 2 {
+			t.Errorf("expected re authorization to happen twice got %d", n)
+		}
+
+		// We should only try try to reconnect once after token error.
+		if expect, got := 2, dials.Load().(int); got != expect {
+			t.Errorf("expected %v got %v", expect, got)
+		}
+	})
+
+}
+
+type closeConn struct {
+	proto.Conn
+	closed int
+}
+
+func (c *closeConn) Close() error {
+	c.closed++
+	return c.Conn.Close()
+}
+
+type noopConn struct {
+	ch chan struct{}
+}
+
+func (noopConn) Send(*proto.ProtocolMessage) error {
+	return nil
+}
+
+func (n *noopConn) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
+	n.ch <- struct{}{}
+	return &proto.ProtocolMessage{}, nil
+}
+func (noopConn) Close() error { return nil }
+
+func TestRealtimeConn_RTN14g(t *testing.T) {
+	t.Parallel()
+	t.Run("Non RTN14b error", func(t *testing.T) {
+		in := make(chan *proto.ProtocolMessage, 1)
+		out := make(chan *proto.ProtocolMessage, 16)
+		var ls *closeConn
+		c, _ := ably.NewRealtime(
+			ably.WithToken("fake:token"),
+			ably.WithAutoConnect(false),
+			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+				w, err := ablytest.MessagePipe(in, out)(protocol, u, timeout)
+				if err != nil {
+					return nil, err
+				}
+				ls = &closeConn{Conn: w}
+				return ls, nil
+			}))
+		badReqErr := &proto.ErrorInfo{
+			StatusCode: http.StatusBadRequest,
+		}
+		in <- &proto.ProtocolMessage{
+			Action:            proto.ActionError,
+			ConnectionDetails: &proto.ConnectionDetails{},
+			Error:             badReqErr,
+		}
+		change := make(ably.ConnStateChanges, 1)
+		c.Connection.OnAll(change.Receive)
+		c.Connect()
+		var state ably.ConnectionStateChange
+		ablytest.Instantly.Recv(t, nil, change, t.Fatalf) // Skip CONNECTING
+		ablytest.Instantly.Recv(t, &state, change, t.Fatalf)
+		if expect, got := ably.ConnectionStateFailed, state.Current; expect != got {
+			t.Errorf("expected %v got %v", expect, got)
+		}
+		if expect, got := badReqErr, c.Connection.ErrorReason(); got.StatusCode != expect.StatusCode {
+			t.Errorf("expected %v got %v", expect, got)
+		}
+		// we make sure the connection is closed
+		if expect, got := 1, ls.closed; got != expect {
+			t.Errorf("expected %v got %v", expect, got)
+		}
+	})
+}
+
+func TestRealtimeConn_RTN14e(t *testing.T) {
+	t.Parallel()
+	ttl := 10 * time.Millisecond
+	disconnTTL := ttl * 2
+	suspendTTL := ttl / 2
+	c, _ := ably.NewRealtime(
+		ably.WithToken("fake:token"),
+		ably.WithAutoConnect(false),
+		ably.WithConnectionStateTTL(ttl),
+		ably.WithSuspendedRetryTimeout(suspendTTL),
+		ably.WithDisconnectedRetryTimeout(disconnTTL),
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+			return nil, context.DeadlineExceeded
+		}))
+	defer c.Close()
+	changes := make(ably.ConnStateChanges, 2)
+	off := c.Connection.On(ably.ConnectionEventSuspended, changes.Receive)
+	defer off()
+	c.Connect()
+	var state ably.ConnectionStateChange
+	ablytest.Soon.Recv(t, &state, changes, t.Fatalf)
+	if state.RetryIn != suspendTTL {
+		t.Fatalf("expected retry to be in %v got %v ", suspendTTL, state.RetryIn)
+	}
+	reason := fmt.Sprintf(ably.ConnectionStateTTLErrFmt, ttl)
+	if !strings.Contains(state.Reason.Error(), reason) {
+		t.Errorf("expected %v\n to contain %s", state.Reason, reason)
+	}
+	// make sure we are from DISCONNECTED => SUSPENDED
+	if expect, got := ably.ConnectionStateDisconnected, state.Previous; got != expect {
+		t.Fatalf("expected transitioning from %v got %v", expect, got)
+	}
+	state = ably.ConnectionStateChange{}
+	ablytest.Soon.Recv(t, &state, changes, t.Fatalf)
+
+	// in suspend retries we move from
+	//  SUSPENDED => CONNECTING => SUSPENDED ...
+	if expect, got := ably.ConnectionStateConnecting, state.Previous; got != expect {
+		t.Fatalf("expected transitioning from %v got %v", expect, got)
+	}
+}
diff --git a/ably/realtime_conn_test.go b/ably/realtime_conn_test.go
index 9b59d6f..0cc37cf 100644
--- a/ably/realtime_conn_test.go
+++ b/ably/realtime_conn_test.go
@@ -122,9 +122,6 @@ func TestRealtimeConn_AuthError(t *testing.T) {
 	if err = ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err == nil {
 		t.Fatal("Connect(): want err != nil")
 	}
-	if state := client.Connection.State(); state != ably.ConnectionStateFailed {
-		t.Fatalf("want state=%s; got %s", ably.ConnectionStateFailed, state)
-	}
 }
 
 func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
diff --git a/ably/realtime_presence_test.go b/ably/realtime_presence_test.go
index 64ddedd..3238621 100644
--- a/ably/realtime_presence_test.go
+++ b/ably/realtime_presence_test.go
@@ -44,14 +44,13 @@ func TestRealtimePresence_Sync(t *testing.T) {
 	t.Parallel()
 	app, client := ablytest.NewRealtime(nil...)
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
+	err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait()
 
 	members, err := client.Channels.Get("persisted:presence_fixtures").Presence.Get(context.Background())
 	if err != nil {
 		t.Fatal(err)
 	}
-	if err != nil {
-		t.Fatal(err)
-	}
+
 	if err = contains(members, fixtureMembers...); err != nil {
 		t.Fatal(err)
 	}
@@ -64,7 +63,18 @@ func TestRealtimePresence_Sync250(t *testing.T) {
 	client2 := app.NewRealtime(nil...)
 	client3 := app.NewRealtime(nil...)
 	defer safeclose(t, ablytest.FullRealtimeCloser(client2), ablytest.FullRealtimeCloser(client3))
-
+	err := ablytest.ConnWaiter(client1, client1.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = ablytest.ConnWaiter(client2, client2.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = ablytest.ConnWaiter(client3, client3.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
 	channel1 := client1.Channels.Get("sync250")
 	if err := channel1.Attach(context.Background()); err != nil {
 		t.Fatal(err)