Repository: ably/ably-go
Pull request #181 (2020-08-02 18:48:33)

RTN15g: Don't attempt resume after server has discarded state
[RTN15g](https://docs.ably.io/client-lib-development-guide/features/#RTN15g)


Patch:
diff --git a/ably/ably.go b/ably/ably.go
index 9a2e35a..94dd404 100644
--- a/ably/ably.go
+++ b/ably/ably.go
@@ -5,6 +5,7 @@ package ably
 import (
 	"crypto/rand"
 	"encoding/hex"
+	"time"
 )
 
 func min(i, j int) int {
@@ -44,3 +45,8 @@ func randomString(n int) string {
 	rand.Read(p)
 	return hex.EncodeToString(p)[:n]
 }
+
+// unixMilli returns the given time as a timestamp in milliseconds since epoch.
+func unixMilli(t time.Time) int64 {
+	return t.UnixNano() / int64(time.Millisecond)
+}
diff --git a/ably/auth.go b/ably/auth.go
index ad5cb5a..50a80de 100644
--- a/ably/auth.go
+++ b/ably/auth.go
@@ -72,9 +72,6 @@ type Auth struct {
 
 	// ServerTimeHandler when provided this will be used to query server time.
 	serverTimeHandler func() (time.Time, error)
-
-	// This provides a function that returns the current time.
-	now func() time.Time
 }
 
 func newAuth(client *REST) (*Auth, error) {
@@ -281,7 +278,7 @@ func (a *Auth) Authorize(params *TokenParams, setOpts AuthOptions) (*TokenDetail
 
 func (a *Auth) authorize(params *TokenParams, opts *authOptions, force bool) (*TokenDetails, error) {
 	switch tok := a.token(); {
-	case tok != nil && !force && (tok.Expires == 0 || !tok.Expired()):
+	case tok != nil && !force && (tok.Expires == 0 || !tok.expired(a.opts().Now())):
 		return tok, nil
 	case params != nil && params.ClientID == "":
 		params.ClientID = a.clientID
@@ -342,19 +339,14 @@ func (a *Auth) setDefaults(opts *authOptions, req *TokenRequest) error {
 		if err != nil {
 			return err
 		}
-		req.Timestamp = Time(ts)
+		req.Timestamp = unixMilli(ts)
 	}
 	return nil
 }
 
 //Timestamp returns the timestamp to be used in authorization request.
 func (a *Auth) timestamp(query bool) (time.Time, error) {
-	var now time.Time
-	if a.now != nil {
-		now = a.now()
-	} else {
-		now = time.Now()
-	}
+	now := a.client.opts.Now()
 	if !query {
 		return now, nil
 	}
diff --git a/ably/options.go b/ably/options.go
index 5eb34ad..e41cf90 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -34,6 +34,7 @@ var defaultOptions = clientOptions{
 	IdempotentRestPublishing: false,
 	Port:                     80,
 	TLSPort:                  443,
+	Now:                      time.Now,
 }
 
 func DefaultFallbackHosts() []string {
@@ -220,6 +221,9 @@ type clientOptions struct {
 
 	//When provided this will be used on every request.
 	Trace *httptrace.ClientTrace
+
+	// Now returns the time the library should take as current.
+	Now func() time.Time
 }
 
 func (opts *clientOptions) timeoutConnect() time.Duration {
@@ -318,38 +322,21 @@ func (opts *clientOptions) idempotentRestPublishing() bool {
 	return opts.IdempotentRestPublishing
 }
 
-// Time returns the given time as a timestamp in milliseconds since epoch.
-func Time(t time.Time) int64 {
-	return t.UnixNano() / int64(time.Millisecond)
-}
-
-// TimeNow returns current time as a timestamp in milliseconds since epoch.
-func TimeNow() int64 {
-	return Time(time.Now())
-}
-
-// Duration returns converts the given duration to milliseconds.
-func Duration(d time.Duration) int64 {
-	return int64(d / time.Millisecond)
-}
-
-// This needs to use a timestamp in millisecond
-// Use the previous function to generate them from a time.Time struct.
 type ScopeParams struct {
-	Start int64
-	End   int64
+	Start time.Time
+	End   time.Time
 	Unit  string
 }
 
-func (s *ScopeParams) EncodeValues(out *url.Values) error {
-	if s.Start != 0 && s.End != 0 && s.Start > s.End {
-		return fmt.Errorf("start must be before end")
+func (s ScopeParams) EncodeValues(out *url.Values) error {
+	if !s.Start.IsZero() && !s.End.IsZero() && s.Start.After(s.End) {
+		return fmt.Errorf("start mzust be before end")
 	}
-	if s.Start != 0 {
-		out.Set("start", strconv.FormatInt(s.Start, 10))
+	if !s.Start.IsZero() {
+		out.Set("start", strconv.FormatInt(unixMilli(s.Start), 10))
 	}
-	if s.End != 0 {
-		out.Set("end", strconv.FormatInt(s.End, 10))
+	if !s.End.IsZero() {
+		out.Set("end", strconv.FormatInt(unixMilli(s.End), 10))
 	}
 	if s.Unit != "" {
 		out.Set("unit", s.Unit)
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index e90a515..1909788 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -64,15 +64,14 @@ func (c *Realtime) onChannelMsg(msg *proto.ProtocolMessage) {
 	c.Channels.Get(msg.Channel).notify(msg)
 }
 
-func (c *Realtime) onReconnectMsg(msg *proto.ProtocolMessage) {
+func (c *Realtime) onReconnectMsg(msg *proto.ProtocolMessage, isNewID bool) {
 	switch msg.Action {
 	case proto.ActionConnected:
-		if msg.Error != nil {
-			// (RTN15c3)
+		if msg.Error != nil || // RTN15c3
+			isNewID { // RTN15g3
 			for _, ch := range c.Channels.All() {
 				switch ch.State() {
-				case StateConnSuspended:
-					ch.attach(false)
+				// TODO: SUSPENDED
 				case StateChanAttaching, StateChanAttached:
 					ch.mayAttach(false, false)
 				}
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index 313322f..62c1fe8 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -56,7 +56,7 @@ type connCallbacks struct {
 	// move this up because some implementation details for (RTN15c) requires
 	// access to Channels and we dont have it here so we let RealtimeClient do the
 	// work.
-	onReconnectMsg func(*proto.ProtocolMessage)
+	onReconnectMsg func(_ *proto.ProtocolMessage, isNewID bool)
 	onStateChange  func(State)
 	// reconnecting tracks if we have issued a reconnection request. If we receive any message
 	// with this set to true then its the first message/response after issuing the
@@ -119,10 +119,23 @@ func (c *Connection) connect(result bool) (Result, error) {
 	return c.connectWith(result, mode)
 }
 
-func (c *Connection) reconnect(result bool) (Result, error) {
+func (c *Connection) reconnect(lastActivityAt time.Time, connDetails *proto.ConnectionDetails, result bool) (Result, error) {
 	c.state.Lock()
-	mode := c.getMode()
+
+	var mode connectionMode
+	if connDetails != nil && c.opts.Now().Sub(lastActivityAt) >= time.Duration(connDetails.ConnectionStateTTL+connDetails.MaxIdleInterval) {
+		// RTN15g
+		c.msgSerial = 0
+		c.key = ""
+		// c.id isn't cleared since it's used later to determine if the
+		// reconnection resulted in a new transport-level connection.
+		mode = normalMode
+	} else {
+		mode = c.getMode()
+	}
+
 	c.state.Unlock()
+
 	r, err := c.connectWith(result, mode)
 	if err != nil {
 		return nil, err
@@ -133,6 +146,7 @@ func (c *Connection) reconnect(result bool) (Result, error) {
 	c.state.Lock()
 	c.reconnecting = true
 	c.state.Unlock()
+
 	return r, nil
 }
 
@@ -148,7 +162,7 @@ func (c *Connection) getMode() connectionMode {
 
 func (c *Connection) params(mode connectionMode) (url.Values, error) {
 	query := url.Values{
-		"timestamp": []string{strconv.FormatInt(TimeNow(), 10)},
+		"timestamp": []string{strconv.FormatInt(unixMilli(c.opts.Now()), 10)},
 		"echo":      []string{"true"},
 		"format":    []string{"msgpack"},
 	}
@@ -479,11 +493,14 @@ func (c *Connection) setSerial(serial int64) {
 }
 
 func (c *Connection) eventloop() {
-	var receiveTimeout time.Duration
+	var lastActivityAt time.Time
+	var connDetails *proto.ConnectionDetails
 	for c.lockCanReceiveMessages() {
 		var deadline time.Time
-		if receiveTimeout != 0 {
-			deadline = time.Now().Add(receiveTimeout) // RTN23a
+		if connDetails != nil {
+			maxIdleInterval := time.Duration(connDetails.MaxIdleInterval)
+			receiveTimeout := c.opts.realtimeRequestTimeout() + maxIdleInterval // RTN23a
+			deadline = c.opts.Now().Add(receiveTimeout)                         // RTNf23a
 		}
 		msg, err := c.conn.Receive(deadline)
 		if err != nil {
@@ -495,9 +512,10 @@ func (c *Connection) eventloop() {
 
 			c.setState(StateConnDisconnected, err)
 			c.state.Unlock()
-			c.reconnect(false)
+			c.reconnect(lastActivityAt, connDetails, false)
 			return
 		}
+		lastActivityAt = c.opts.Now()
 		if msg.ConnectionSerial != 0 {
 			c.state.Lock()
 			c.setSerial(msg.ConnectionSerial)
@@ -527,7 +545,7 @@ func (c *Connection) eventloop() {
 					// TODO: (gernest) implement (RTN15h) This can be done as a separate task?
 				} else {
 					// (RTN15c4)
-					c.callbacks.onReconnectMsg(msg)
+					c.callbacks.onReconnectMsg(msg, false)
 				}
 			}
 			c.setState(StateConnFailed, newErrorProto(msg.Error))
@@ -537,22 +555,20 @@ func (c *Connection) eventloop() {
 				c.conn.Close()
 			}
 		case proto.ActionConnected:
-			c.auth.updateClientID(msg.ConnectionDetails.ClientID)
 			c.state.Lock()
 			// we need to get this before we set c.key so as to be sure if we were
 			// resuming or recovering the connection.
 			mode := c.getMode()
 			c.state.Unlock()
 			if msg.ConnectionDetails != nil {
+				connDetails = msg.ConnectionDetails
+
 				c.state.Lock()
-				c.key = msg.ConnectionDetails.ConnectionKey //(RTN15e) (RTN16d)
+				c.key = connDetails.ConnectionKey //(RTN15e) (RTN16d)
 				c.state.Unlock()
 
 				// Spec RSA7b3, RSA7b4, RSA12a
-				c.auth.updateClientID(msg.ConnectionDetails.ClientID)
-
-				maxIdleInterval := time.Duration(msg.ConnectionDetails.MaxIdleInterval) * time.Millisecond
-				receiveTimeout = c.opts.realtimeRequestTimeout() + maxIdleInterval // RTN23a
+				c.auth.updateClientID(connDetails.ClientID)
 			}
 			c.state.Lock()
 			reconnecting := c.reconnecting
@@ -572,7 +588,7 @@ func (c *Connection) eventloop() {
 					// we are calling this outside of locks to avoid deadlock because in the
 					// RealtimeClient client where this callback is implemented we do some ops
 					// with this Conn where we re acquire Conn.state.Lock again.
-					c.callbacks.onReconnectMsg(msg)
+					c.callbacks.onReconnectMsg(msg, true)
 				}
 			} else {
 				// preserve old behavior.
diff --git a/ably/token.go b/ably/token.go
index 75ba01a..7e4b906 100644
--- a/ably/token.go
+++ b/ably/token.go
@@ -133,9 +133,8 @@ func (tok *TokenDetails) Capability() Capability {
 	return c
 }
 
-// Expired
-func (tok *TokenDetails) Expired() bool {
-	return tok.Expires != 0 && tok.Expires <= TimeNow()
+func (tok *TokenDetails) expired(now time.Time) bool {
+	return tok.Expires != 0 && tok.Expires <= unixMilli(now)
 }
 
 func (tok *TokenDetails) IssueTime() time.Time {


Test patch:
diff --git a/ably/ablytest/recorders.go b/ably/ablytest/recorders.go
index 45e4f80..4a3f471 100644
--- a/ably/ablytest/recorders.go
+++ b/ably/ablytest/recorders.go
@@ -310,7 +310,10 @@ func (pc pipeConn) Send(msg *proto.ProtocolMessage) error {
 
 func (pc pipeConn) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
 	select {
-	case m := <-pc.in:
+	case m, ok := <-pc.in:
+		if !ok {
+			return nil, io.EOF
+		}
 		return m, nil
 	case <-time.After(time.Until(deadline)):
 		return nil, errTimeout{}
diff --git a/ably/auth_test.go b/ably/auth_test.go
index 8ab022f..b8f231a 100644
--- a/ably/auth_test.go
+++ b/ably/auth_test.go
@@ -172,7 +172,7 @@ func TestAuth_Authorise(t *testing.T) {
 	defer safeclose(t, app)
 
 	params := &ably.TokenParams{
-		TTL: ably.Duration(time.Second),
+		TTL: time.Second.Milliseconds(),
 	}
 	_, err := client.Auth.Authorise(params, nil)
 	if err != nil {
@@ -196,10 +196,12 @@ func TestAuth_TimestampRSA10k(t *testing.T) {
 	}
 
 	t.Run("must use local time when UseQueryTime is false", func(ts *testing.T) {
-		a := &ably.Auth{}
-		a.SetNowFunc(func() time.Time {
-			return now
-		})
+		rest, _ := ably.NewREST(ably.ClientOptions{}.
+			Key("fake:key").
+			Now(func() time.Time {
+				return now
+			}))
+		a := rest.Auth
 		a.SetServerTimeFunc(func() (time.Time, error) {
 			return now.Add(time.Minute), nil
 		})
@@ -212,10 +214,12 @@ func TestAuth_TimestampRSA10k(t *testing.T) {
 		}
 	})
 	t.Run("must use server time when UseQueryTime is true", func(ts *testing.T) {
-		a := &ably.Auth{}
-		a.SetNowFunc(func() time.Time {
-			return now
-		})
+		rest, _ := ably.NewREST(ably.ClientOptions{}.
+			Key("fake:key").
+			Now(func() time.Time {
+				return now
+			}))
+		a := rest.Auth
 		a.SetServerTimeFunc(func() (time.Time, error) {
 			return now.Add(time.Minute), nil
 		})
@@ -229,10 +233,13 @@ func TestAuth_TimestampRSA10k(t *testing.T) {
 		}
 	})
 	t.Run("must use server time offset ", func(ts *testing.T) {
-		a := &ably.Auth{}
-		a.SetNowFunc(func() time.Time {
-			return now
-		})
+		now := now
+		rest, _ := ably.NewREST(ably.ClientOptions{}.
+			Key("fake:key").
+			Now(func() time.Time {
+				return now
+			}))
+		a := rest.Auth
 		a.SetServerTimeFunc(func() (time.Time, error) {
 			return now.Add(time.Minute), nil
 		})
@@ -245,9 +252,7 @@ func TestAuth_TimestampRSA10k(t *testing.T) {
 			ts.Errorf("expected %s got %s", serverTime, stamp)
 		}
 
-		a.SetNowFunc(func() time.Time {
-			return now.Add(time.Minute)
-		})
+		now = now.Add(time.Minute)
 		a.SetServerTimeFunc(func() (time.Time, error) {
 			return time.Time{}, errors.New("must not be called")
 		})
@@ -255,7 +260,7 @@ func TestAuth_TimestampRSA10k(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		serverTime = now.Add(time.Minute * 2)
+		serverTime = now.Add(time.Minute)
 		if !stamp.Equal(serverTime) {
 			ts.Errorf("expected %s got %s", serverTime, stamp)
 		}
@@ -271,7 +276,7 @@ func TestAuth_TokenAuth_Renew(t *testing.T) {
 	defer safeclose(t, app)
 
 	params := &ably.TokenParams{
-		TTL: ably.Duration(time.Second),
+		TTL: time.Second.Milliseconds(),
 	}
 	tok, err := client.Auth.Authorize(params, nil)
 	if err != nil {
@@ -608,7 +613,7 @@ func TestAuth_ClientID(t *testing.T) {
 	proxy := ablytest.MustAuthReverseProxy(app.Options(opts))
 	defer safeclose(t, proxy)
 	params := &ably.TokenParams{
-		TTL: ably.Duration(time.Second),
+		TTL: time.Second.Milliseconds(),
 	}
 	opts = ably.ClientOptions{}.
 		AuthURL(proxy.URL("details")).
@@ -713,7 +718,7 @@ func TestAuth_CreateTokenRequest(t *testing.T) {
 		QueryTime(true).
 		Key(app.Key())
 	params := &ably.TokenParams{
-		TTL:           ably.Duration(5 * time.Second),
+		TTL:           (5 * time.Second).Milliseconds(),
 		RawCapability: (ably.Capability{"presence": {"read", "write"}}).Encode(),
 	}
 	t.Run("RSA9h", func(ts *testing.T) {
diff --git a/ably/export_test.go b/ably/export_test.go
index 2f09c9c..24c2e70 100644
--- a/ably/export_test.go
+++ b/ably/export_test.go
@@ -53,10 +53,6 @@ func (a *Auth) Timestamp(query bool) (time.Time, error) {
 	return a.timestamp(query)
 }
 
-func (a *Auth) SetNowFunc(now func() time.Time) {
-	a.now = now
-}
-
 func (a *Auth) SetServerTimeFunc(st func() (time.Time, error)) {
 	a.serverTimeHandler = st
 }
@@ -129,6 +125,12 @@ func (os ClientOptions) Trace(trace *httptrace.ClientTrace) ClientOptions {
 	})
 }
 
+func (os ClientOptions) Now(now func() time.Time) ClientOptions {
+	return append(os, func(os *clientOptions) {
+		os.Now = now
+	})
+}
+
 func (os ClientOptions) ApplyWithDefaults() *clientOptions {
 	return os.applyWithDefaults()
 }
diff --git a/ably/options_test.go b/ably/options_test.go
index 1578eb1..47e8862 100644
--- a/ably/options_test.go
+++ b/ably/options_test.go
@@ -3,6 +3,7 @@ package ably_test
 import (
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/ably/ably-go/ably"
 )
@@ -21,8 +22,8 @@ func TestScopeParams(t *testing.T) {
 	t.Parallel()
 	t.Run("must error when given invalid range", func(ts *testing.T) {
 		params := ably.ScopeParams{
-			Start: 123,
-			End:   122,
+			Start: time.Unix(0, 0).Add(123 * time.Millisecond),
+			End:   time.Unix(0, 0).Add(122 * time.Millisecond),
 		}
 		err := params.EncodeValues(nil)
 		if err == nil {
@@ -32,8 +33,8 @@ func TestScopeParams(t *testing.T) {
 
 	t.Run("must set url values", func(ts *testing.T) {
 		params := ably.ScopeParams{
-			Start: 122,
-			End:   123,
+			Start: time.Unix(0, 0).Add(122 * time.Millisecond),
+			End:   time.Unix(0, 0).Add(123 * time.Millisecond),
 		}
 		u := make(url.Values)
 		err := params.EncodeValues(&u)
@@ -71,8 +72,8 @@ func TestPaginateParams(t *testing.T) {
 			Limit:     1,
 			Direction: "backwards",
 			ScopeParams: ably.ScopeParams{
-				Start: 123,
-				End:   124,
+				Start: time.Unix(0, 0).Add(123 * time.Millisecond),
+				End:   time.Unix(0, 0).Add(124 * time.Millisecond),
 				Unit:  "hello",
 			},
 		}
@@ -103,7 +104,7 @@ func TestPaginateParams(t *testing.T) {
 	t.Run("with a value for ScopeParams", func(ts *testing.T) {
 		values := make(url.Values)
 		params := ably.PaginateParams{}
-		params.Start = 123
+		params.Start = time.Unix(0, 0).Add(123 * time.Millisecond)
 		err := params.EncodeValues(&values)
 		if err != nil {
 			ts.Fatal(err)
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index 4f9e6bf..419559f 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -971,6 +972,169 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 	}
 }
 
+func TestRealtimeConn_RTN15g_NewConnectionOnStateLost(t *testing.T) {
+	t.Parallel()
+
+	out := make(chan *proto.ProtocolMessage, 16)
+
+	now, setNow := func() (func() time.Time, func(time.Time)) {
+		now := time.Now()
+
+		var mtx sync.Mutex
+		return func() time.Time {
+				mtx.Lock()
+				defer mtx.Unlock()
+				return now
+			}, func(t time.Time) {
+				mtx.Lock()
+				defer mtx.Unlock()
+				now = t
+			}
+	}()
+
+	connDetails := proto.ConnectionDetails{
+		ConnectionKey:      "foo",
+		ConnectionStateTTL: proto.DurationFromMsecs(time.Minute * 2),
+		MaxIdleInterval:    proto.DurationFromMsecs(time.Second),
+	}
+
+	dials := make(chan *url.URL, 1)
+	connIDs := make(chan string, 1)
+	var breakConn func()
+	var in chan *proto.ProtocolMessage
+
+	c, _ := ably.NewRealtime(ably.ClientOptions{}.
+		AutoConnect(false).
+		Token("fake:token").
+		Now(now).
+		Dial(func(p string, u *url.URL) (proto.Conn, error) {
+			in = make(chan *proto.ProtocolMessage, 1)
+			in <- &proto.ProtocolMessage{
+				Action:            proto.ActionConnected,
+				ConnectionID:      <-connIDs,
+				ConnectionDetails: &connDetails,
+			}
+			breakConn = func() { close(in) }
+			dials <- u
+			return ablytest.MessagePipe(in, out)(p, u)
+		}))
+
+	connIDs <- "conn-1"
+	err := ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ablytest.Instantly.Recv(t, nil, dials, t.Fatalf) // discard first URL; we're interested in reconnections
+
+	// Get channels to ATTACHING, ATTACHED and DETACHED. (TODO: SUSPENDED)
+
+	attaching := c.Channels.Get("attaching")
+	_, err = attaching.Attach()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg := <-out; msg.Action != proto.ActionAttach {
+		t.Fatalf("expected ATTACH, got %v", msg)
+	}
+
+	attached := c.Channels.Get("attached")
+	attachWaiter, err := attached.Attach()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg := <-out; msg.Action != proto.ActionAttach {
+		t.Fatalf("expected ATTACH, got %v", msg)
+	}
+	in <- &proto.ProtocolMessage{
+		Action:  proto.ActionAttached,
+		Channel: "attached",
+	}
+	ablytest.Wait(attachWaiter, err)
+
+	detached := c.Channels.Get("detached")
+	attachWaiter, err = detached.Attach()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg := <-out; msg.Action != proto.ActionAttach {
+		t.Fatalf("expected ATTACH, got %v", msg)
+	}
+	in <- &proto.ProtocolMessage{
+		Action:  proto.ActionAttached,
+		Channel: "detached",
+	}
+	ablytest.Wait(attachWaiter, err)
+	detachWaiter, err := detached.Detach()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg := <-out; msg.Action != proto.ActionDetach {
+		t.Fatalf("expected DETACH, got %v", msg)
+	}
+	in <- &proto.ProtocolMessage{
+		Action:  proto.ActionDetached,
+		Channel: "detached",
+	}
+	ablytest.Wait(detachWaiter, err)
+
+	// Simulate a broken connection. Before that, set the current time to a point
+	// in the future just before connectionStateTTL + maxIdleInterval. So we still
+	// attempt a resume.
+
+	discardStateTTL := time.Duration(connDetails.ConnectionStateTTL + connDetails.MaxIdleInterval)
+
+	setNow(now().Add(discardStateTTL - 1))
+
+	connected := make(chan struct{})
+	c.Connection.Once(ably.ConnectionEventConnected, func(ably.ConnectionStateChange) {
+		close(connected)
+	})
+
+	breakConn()
+
+	connIDs <- "conn-1" // Same connection ID so the resume "succeeds".
+	var dialed *url.URL
+	ablytest.Instantly.Recv(t, &dialed, dials, t.Fatalf)
+	if resume := dialed.Query().Get("resume"); resume == "" {
+		t.Fatalf("expected a resume key; got %v", dialed)
+	}
+
+	// Now do the same, but past connectionStateTTL + maxIdleInterval. This
+	// should make a fresh connection.
+
+	ablytest.Instantly.Recv(t, nil, connected, t.Fatalf) // wait for CONNECTED before disconnecting again
+
+	setNow(now().Add(discardStateTTL + 1))
+	breakConn()
+
+	connIDs <- "conn-2"
+	ablytest.Instantly.Recv(t, &dialed, dials, t.Fatalf)
+	if resume := dialed.Query().Get("resume"); resume != "" {
+		t.Fatalf("didn't expect a resume key; got %v", dialed)
+	}
+	if recover := dialed.Query().Get("recover"); recover != "" {
+		t.Fatalf("didn't expect a recover key; got %v", dialed)
+	}
+
+	// RTN15g3: Expect the previously attaching and attached channels to be
+	// attached again.
+	attachExpected := map[string]struct{}{
+		"attaching": {},
+		"attached":  {},
+	}
+	for len(attachExpected) > 0 {
+		var msg *proto.ProtocolMessage
+		ablytest.Instantly.Recv(t, &msg, out, t.Fatalf)
+		_, ok := attachExpected[msg.Channel]
+		if !ok {
+			t.Fatalf("ATTACH sent for unexpected or already attaching channel %q", msg.Channel)
+		}
+		delete(attachExpected, msg.Channel)
+	}
+	ablytest.Instantly.NoRecv(t, nil, out, t.Fatalf)
+}
+
 func TestRealtimeConn_RTN15i_OnErrorWhenConnected(t *testing.T) {
 	t.Parallel()
 
diff --git a/ably/rest_client_test.go b/ably/rest_client_test.go
index ad4d443..1151927 100644
--- a/ably/rest_client_test.go
+++ b/ably/rest_client_test.go
@@ -200,7 +200,7 @@ func TestRestClient(t *testing.T) {
 					page, err := client.Stats(&ably.PaginateParams{
 						Limit: 1,
 						ScopeParams: ably.ScopeParams{
-							Start: ably.Time(longAgo),
+							Start: longAgo,
 							Unit:  proto.StatGranularityMinute,
 						},
 					})
diff --git a/ably/rest_presence_test.go b/ably/rest_presence_test.go
index aee4e98..36cdc68 100644
--- a/ably/rest_presence_test.go
+++ b/ably/rest_presence_test.go
@@ -82,8 +82,8 @@ func TestChannel_Presence(t *testing.T) {
 		ts.Run("with start and end time", func(ts *testing.T) {
 			params := &ably.PaginateParams{
 				ScopeParams: ably.ScopeParams{
-					Start: ably.Time(time.Now().Add(-24 * time.Hour)),
-					End:   ably.Time(time.Now()),
+					Start: time.Now().Add(-24 * time.Hour),
+					End:   time.Now(),
 				},
 			}
 			page, err := presence.History(params)