Merge lp:~pedronis/ubuntu-push/kick-out-replaced into lp:ubuntu-push

[Samuele Pedroni]

explicitly kick out superseded sessions

explicitly kick out superseded sessions

Unmerged revisions

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-04-04 12:28:40 +0000
+++ client/session/session.go	2014-04-11 09:14:12 +0000
@@ -49,6 +49,7 @@
 	Type string `json:"T"`
 	protocol.BroadcastMsg
 	protocol.NotificationsMsg
+	protocol.ConnBrokenMsg
 }
 
 // parseServerAddrSpec recognizes whether spec is a HTTP URL to get
@@ -176,7 +177,7 @@
 // getHosts sets deliveryHosts possibly querying a remote endpoint
 func (sess *ClientSession) getHosts() error {
 	if sess.getHost != nil {
-		if sess.timeSince(sess.deliveryHostsTimestamp) < sess.HostsCachingExpiryTime {
+		if sess.deliveryHosts != nil && sess.timeSince(sess.deliveryHostsTimestamp) < sess.HostsCachingExpiryTime {
 			return nil
 		}
 		hosts, err := sess.getHost.Get()
@@ -193,6 +194,10 @@
 	return nil
 }
 
+func (sess *ClientSession) resetHosts() {
+	sess.deliveryHosts = nil
+}
+
 // startConnectionAttempt/nextHostToTry help connect iterating over candidate hosts
 
 func (sess *ClientSession) startConnectionAttempt() {
@@ -338,6 +343,19 @@
 	return nil
 }
 
+// handle "connbroken" messages
+func (sess *ClientSession) handleConnBroken(connBroken *serverMsg) error {
+	sess.setState(Error)
+	reason := connBroken.Reason
+	err := fmt.Errorf("server broke connection: %s", reason)
+	sess.Log.Errorf("%s", err)
+	switch reason {
+	case protocol.BrokenHostMismatch:
+		sess.resetHosts()
+	}
+	return err
+}
+
 // loop runs the session with the server, emits a stream of events.
 func (sess *ClientSession) loop() error {
 	var err error
@@ -356,6 +374,8 @@
 			err = sess.handlePing()
 		case "broadcast":
 			err = sess.handleBroadcast(&recv)
+		case "connbroken":
+			err = sess.handleConnBroken(&recv)
 		}
 		if err != nil {
 			return err

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-04-03 20:56:25 +0000
+++ client/session/session_test.go	2014-04-11 09:14:12 +0000
@@ -317,6 +317,29 @@
 	c.Check(sess.deliveryHosts, DeepEquals, []string{"baz:443"})
 }
 
+func (cs *clientSessionSuite) TestGetHostsRemoteCachingReset(c *C) {
+	hostGetter := &testHostGetter{[]string{"foo:443", "bar:443"}, nil}
+	sess := &ClientSession{
+		getHost: hostGetter,
+		ClientSessionConfig: ClientSessionConfig{
+			HostsCachingExpiryTime: 2 * time.Hour,
+		},
+		timeSince: time.Since,
+	}
+	err := sess.getHosts()
+	c.Assert(err, IsNil)
+	hostGetter.hosts = []string{"baz:443"}
+	// cached
+	err = sess.getHosts()
+	c.Assert(err, IsNil)
+	c.Check(sess.deliveryHosts, DeepEquals, []string{"foo:443", "bar:443"})
+	// reset
+	sess.resetHosts()
+	err = sess.getHosts()
+	c.Assert(err, IsNil)
+	c.Check(sess.deliveryHosts, DeepEquals, []string{"baz:443"})
+}
+
 /****************************************************************
   startConnectionAttempt()/nextHostToTry()/started tests
 ****************************************************************/
@@ -587,7 +610,7 @@
 				json.RawMessage("false"), // shouldn't happen but robust
 				json.RawMessage(`{"img1/m1":[102,"tubular"]}`),
 			},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
@@ -618,7 +641,7 @@
 			ChanId:   "0",
 			TopLevel: 2,
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	failure := errors.New("ACK ACK ACK")
@@ -635,7 +658,7 @@
 			ChanId:   "something awful",
 			TopLevel: 2,
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	c.Check(takeNext(s.downCh), Equals, protocol.AckMsg{"ack"})
 	s.upCh <- nil // ack ok
@@ -652,7 +675,7 @@
 			ChanId:   "0",
 			TopLevel: 2,
 			Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
-		}, protocol.NotificationsMsg{}}
+		}, protocol.NotificationsMsg{}, protocol.ConnBrokenMsg{}}
 	go func() { s.errCh <- s.sess.handleBroadcast(&msg) }()
 	s.upCh <- nil // ack ok
 	// start returns with error
@@ -665,6 +688,37 @@
 }
 
 /****************************************************************
+  handleConnBroken() tests
+****************************************************************/
+
+func (s *msgSuite) TestHandleConnBrokenUnkwown(c *C) {
+	msg := serverMsg{"connbroken",
+		protocol.BroadcastMsg{}, protocol.NotificationsMsg{},
+		protocol.ConnBrokenMsg{
+			Reason: "REASON",
+		},
+	}
+	go func() { s.errCh <- s.sess.handleConnBroken(&msg) }()
+	c.Check(<-s.errCh, ErrorMatches, "server broke connection: REASON")
+	c.Check(s.sess.State(), Equals, Error)
+}
+
+func (s *msgSuite) TestHandleConnBrokenHostMismatch(c *C) {
+	msg := serverMsg{"connbroken",
+		protocol.BroadcastMsg{}, protocol.NotificationsMsg{},
+		protocol.ConnBrokenMsg{
+			Reason: protocol.BrokenHostMismatch,
+		},
+	}
+	s.sess.deliveryHosts = []string{"foo:443", "bar:443"}
+	go func() { s.errCh <- s.sess.handleConnBroken(&msg) }()
+	c.Check(<-s.errCh, ErrorMatches, "server broke connection: host-mismatch")
+	c.Check(s.sess.State(), Equals, Error)
+	// hosts were reset
+	c.Check(s.sess.deliveryHosts, IsNil)
+}
+
+/****************************************************************
   loop() tests
 ****************************************************************/
 
@@ -728,6 +782,17 @@
 	c.Check(<-s.errCh, Equals, failure)
 }
 
+func (s *loopSuite) TestLoopConnBroken(c *C) {
+	c.Check(s.sess.State(), Equals, Running)
+	broken := protocol.ConnBrokenMsg{
+		Type:   "connbroken",
+		Reason: "REASON",
+	}
+	c.Check(takeNext(s.downCh), Equals, "deadline 1ms")
+	s.upCh <- broken
+	c.Check(<-s.errCh, NotNil)
+}
+
 /****************************************************************
   start() tests
 ****************************************************************/

=== modified file 'protocol/messages.go'
--- protocol/messages.go	2014-03-19 22:31:20 +0000
+++ protocol/messages.go	2014-04-11 09:14:12 +0000
@@ -49,6 +49,27 @@
 	PingInterval string
 }
 
+// SplittableMsg are messages that may require and are capable of splitting.
+type SplittableMsg interface {
+	Split() (done bool)
+}
+
+// CONNBROKEN message, server side is breaking the connection for reason.
+type ConnBrokenMsg struct {
+	Type string `json:"T"`
+	// reason
+	Reason string
+}
+
+func (m *ConnBrokenMsg) Split() bool {
+	return true
+}
+
+// CONNBROKEN reasons
+const (
+	BrokenHostMismatch = "host-mismatch"
+)
+
 // PING/PONG messages
 type PingPongMsg struct {
 	Type string `json:"T"`
@@ -56,11 +77,6 @@
 
 const maxPayloadSize = 62 * 1024
 
-// SplittableMsg are messages that may require and are capable of splitting.
-type SplittableMsg interface {
-	Split() (done bool)
-}
-
 // BROADCAST messages
 type BroadcastMsg struct {
 	Type      string `json:"T"`

=== modified file 'protocol/messages_test.go'
--- protocol/messages_test.go	2014-02-26 16:04:57 +0000
+++ protocol/messages_test.go	2014-04-11 09:14:12 +0000
@@ -103,3 +103,7 @@
 	b.Reset()
 	c.Check(b.splitting, Equals, 0)
 }
+
+func (s *messagesSuite) TestSplitConnBrokenMsg(c *C) {
+	c.Check((&ConnBrokenMsg{}).Split(), Equals, true)
+}

=== modified file 'server/acceptance/suites/broadcast.go'
--- server/acceptance/suites/broadcast.go	2014-04-03 16:47:47 +0000
+++ server/acceptance/suites/broadcast.go	2014-04-11 09:14:12 +0000
@@ -66,8 +66,6 @@
 	})
 	c.Assert(err, IsNil)
 	c.Assert(got, Matches, ".*ok.*")
-	// xxx don't send this one
-	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[]`)
 	c.Check(NextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"img1/m1":20}]`)
 	stop()
 	c.Assert(NextEvent(s.ServerEvents, nil), Matches, `.* ended with:.*EOF`)

=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-04-03 14:31:10 +0000
+++ server/broker/broker.go	2014-04-11 09:14:12 +0000
@@ -19,6 +19,7 @@
 package broker
 
 import (
+	"errors"
 	"fmt"
 
 	"launchpad.net/ubuntu-push/protocol"
@@ -46,6 +47,9 @@
 	Acked(sess BrokerSession, done bool) error
 }
 
+// ErrNop returned by Prepare means nothing to do/send.
+var ErrNop = errors.New("nothing to send")
+
 // LevelsMap is the type for holding channel levels for session.
 type LevelsMap map[store.InternalChannelId]int64
 

=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-04-03 16:00:53 +0000
+++ server/broker/exchanges.go	2014-04-11 09:14:12 +0000
@@ -28,8 +28,9 @@
 
 // Scratch area for exchanges, sessions should hold one of these.
 type ExchangesScratchArea struct {
-	broadcastMsg protocol.BroadcastMsg
-	ackMsg       protocol.AckMsg
+	broadcastMsg  protocol.BroadcastMsg
+	ackMsg        protocol.AckMsg
+	connBrokenMsg protocol.ConnBrokenMsg
 }
 
 // BroadcastExchange leads a session through delivering a BROADCAST.
@@ -42,7 +43,7 @@
 }
 
 // check interface already here
-var _ Exchange = &BroadcastExchange{}
+var _ Exchange = (*BroadcastExchange)(nil)
 
 // Init ensures the BroadcastExchange is fully initialized for the sessions.
 func (sbe *BroadcastExchange) Init() {
@@ -88,14 +89,18 @@
 
 // Prepare session for a BROADCAST.
 func (sbe *BroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
-	scratchArea := sess.ExchangeScratchArea()
-	scratchArea.broadcastMsg.Reset()
-	scratchArea.broadcastMsg.Type = "broadcast"
 	clientLevel := sess.Levels()[sbe.ChanId]
 	payloads := filterByLevel(clientLevel, sbe.TopLevel, sbe.NotificationPayloads)
 	tag := fmt.Sprintf("%s/%s", sess.DeviceImageChannel(), sess.DeviceImageModel())
 	payloads = channelFilter(tag, sbe.ChanId, payloads, sbe.Decoded)
+	if len(payloads) == 0 && sbe.TopLevel >= clientLevel {
+		// empty and don't need to force resync => do nothing
+		return nil, nil, ErrNop
+	}
 
+	scratchArea := sess.ExchangeScratchArea()
+	scratchArea.broadcastMsg.Reset()
+	scratchArea.broadcastMsg.Type = "broadcast"
 	// xxx need an AppId as well, later
 	scratchArea.broadcastMsg.ChanId = store.InternalChannelIdToHex(sbe.ChanId)
 	scratchArea.broadcastMsg.TopLevel = sbe.TopLevel
@@ -113,3 +118,24 @@
 	sess.Levels()[sbe.ChanId] = sbe.TopLevel
 	return nil
 }
+
+// ConnBrokenExchange breaks a session giving a reason.
+type ConnBrokenExchange struct {
+	Reason string
+}
+
+// check interface already here
+var _ Exchange = (*ConnBrokenExchange)(nil)
+
+// Prepare session for a CONNBROKEN.
+func (cbe *ConnBrokenExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
+	scratchArea := sess.ExchangeScratchArea()
+	scratchArea.connBrokenMsg.Type = "connbroken"
+	scratchArea.connBrokenMsg.Reason = cbe.Reason
+	return &scratchArea.connBrokenMsg, nil, nil
+}
+
+// CONNBROKEN isn't acked
+func (cbe *ConnBrokenExchange) Acked(sess BrokerSession, done bool) error {
+	panic("Acked should not get invoked on ConnBrokenExchange")
+}

=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-04-04 09:57:02 +0000
+++ server/broker/exchanges_test.go	2014-04-11 09:14:12 +0000
@@ -81,6 +81,44 @@
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(3))
 }
 
+func (s *exchangesSuite) TestBroadcastExchangeEmpty(c *C) {
+	sess := &testing.TestBrokerSession{
+		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
+		Model:        "m1",
+		ImageChannel: "img1",
+	}
+	exchg := &broker.BroadcastExchange{
+		ChanId:               store.SystemInternalChannelId,
+		TopLevel:             3,
+		NotificationPayloads: []json.RawMessage{},
+	}
+	exchg.Init()
+	outMsg, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, Equals, broker.ErrNop)
+	c.Check(outMsg, IsNil)
+	c.Check(inMsg, IsNil)
+}
+
+func (s *exchangesSuite) TestBroadcastExchangeEmptyButAhead(c *C) {
+	sess := &testing.TestBrokerSession{
+		LevelsMap: broker.LevelsMap(map[store.InternalChannelId]int64{
+			store.SystemInternalChannelId: 10,
+		}),
+		Model:        "m1",
+		ImageChannel: "img1",
+	}
+	exchg := &broker.BroadcastExchange{
+		ChanId:               store.SystemInternalChannelId,
+		TopLevel:             3,
+		NotificationPayloads: []json.RawMessage{},
+	}
+	exchg.Init()
+	outMsg, inMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	c.Check(outMsg, NotNil)
+	c.Check(inMsg, NotNil)
+}
+
 func (s *exchangesSuite) TestBroadcastExchangeReuseVsSplit(c *C) {
 	sess := &testing.TestBrokerSession{
 		LevelsMap:    broker.LevelsMap(map[store.InternalChannelId]int64{}),
@@ -210,3 +248,17 @@
 	c.Assert(err, IsNil)
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(5))
 }
+
+func (s *exchangesSuite) TestConnBrokenExchange(c *C) {
+	sess := &testing.TestBrokerSession{}
+	cbe := &broker.ConnBrokenExchange{"REASON"}
+	outMsg, inMsg, err := cbe.Prepare(sess)
+	c.Assert(err, IsNil)
+	c.Check(inMsg, IsNil) // no answer is expected
+	// check
+	marshalled, err := json.Marshal(outMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"connbroken","Reason":"REASON"}`)
+
+	c.Check(func() { cbe.Acked(nil, true) }, PanicMatches, "Acked should not get invoked on ConnBrokenExchange")
+}

=== modified file 'server/broker/simple/simple.go'
--- server/broker/simple/simple.go	2014-04-03 16:00:53 +0000
+++ server/broker/simple/simple.go	2014-04-11 09:14:12 +0000
@@ -222,6 +222,10 @@
 					delete(b.registry, sess.deviceId)
 				}
 			} else { // register
+				prev := b.registry[sess.deviceId]
+				if prev != nil { // kick it
+					close(prev.exchanges)
+				}
 				b.registry[sess.deviceId] = sess
 				sess.registered = true
 				sess.done <- true

=== modified file 'server/broker/testsuite/suite.go'
--- server/broker/testsuite/suite.go	2014-04-03 16:00:53 +0000
+++ server/broker/testsuite/suite.go	2014-04-11 09:14:12 +0000
@@ -164,6 +164,14 @@
 	c.Assert(err, IsNil)
 	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
 	c.Assert(err, IsNil)
+	checkAndFalse := false
+	// previous session got signaled by closing its channel
+	select {
+	case _, ok := <-sess1.SessionChannel():
+		checkAndFalse = ok == false
+	default:
+	}
+	c.Check(checkAndFalse, Equals, true)
 	c.Assert(s.RevealSession(b, "dev-1"), Equals, sess2)
 	b.Unregister(sess1)
 	// just to make sure the unregister was processed

=== modified file 'server/session/session.go'
--- server/session/session.go	2014-02-10 23:19:08 +0000
+++ server/session/session.go	2014-04-11 09:14:12 +0000
@@ -62,6 +62,9 @@
 	if err != nil {
 		return err
 	}
+	if inMsg == nil { // no answer expected, breaking connection
+		return &broker.ErrAbort{"session broken for reason"}
+	}
 	err = proto.ReadMessage(inMsg)
 	if err != nil {
 		return err
@@ -76,6 +79,7 @@
 	pingTimer := time.NewTimer(pingInterval)
 	intervalStart := time.Now()
 	ch := sess.SessionChannel()
+Loop:
 	for {
 		select {
 		case <-pingTimer.C:
@@ -90,10 +94,17 @@
 				return &broker.ErrAbort{"expected PONG message"}
 			}
 			pingTimer.Reset(pingInterval)
-		case exchg := <-ch:
-			// xxx later can use ch closing for shutdown/reset
+		case exchg, ok := <-ch:
 			pingTimer.Stop()
+			if !ok {
+				return &broker.ErrAbort{"terminated"}
+			}
 			outMsg, inMsg, err := exchg.Prepare(sess)
+			if err == broker.ErrNop { // nothing to do
+				pingTimer.Reset(pingInterval)
+				intervalStart = time.Now()
+				continue Loop
+			}
 			if err != nil {
 				return err
 			}

=== modified file 'server/session/session_test.go'
--- server/session/session_test.go	2014-03-19 23:46:18 +0000
+++ server/session/session_test.go	2014-04-11 09:14:12 +0000
@@ -346,6 +346,42 @@
 	c.Check(err, Equals, io.EOF)
 }
 
+func (s *sessionSuite) TestSessionLoopKick(c *C) {
+	nopTrack := NewTracker(s.testlog)
+	errCh := make(chan error, 1)
+	up := make(chan interface{}, 5)
+	down := make(chan interface{}, 5)
+	tp := &testProtocol{up, down}
+	exchanges := make(chan broker.Exchange, 1)
+	sess := &testing.TestBrokerSession{Exchanges: exchanges}
+	go func() {
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
+	}()
+	close(exchanges)
+	err := <-errCh
+	c.Check(err, DeepEquals, &broker.ErrAbort{"terminated"})
+}
+
+func (s *sessionSuite) TestSessionLoopExchangeErrNop(c *C) {
+	nopTrack := NewTracker(s.testlog)
+	errCh := make(chan error, 1)
+	up := make(chan interface{}, 5)
+	down := make(chan interface{}, 5)
+	tp := &testProtocol{up, down}
+	exchanges := make(chan broker.Exchange, 1)
+	exchanges <- &testExchange{prepErr: broker.ErrNop}
+	sess := &testing.TestBrokerSession{Exchanges: exchanges}
+	go func() {
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
+	}()
+	c.Check(takeNext(down), Equals, "deadline 2ms")
+	c.Check(takeNext(down), DeepEquals, protocol.PingPongMsg{Type: "ping"})
+	up <- nil // no write error
+	up <- io.EOF
+	err := <-errCh
+	c.Check(err, Equals, io.EOF)
+}
+
 func (s *sessionSuite) TestSessionLoopExchangeSplit(c *C) {
 	nopTrack := NewTracker(s.testlog)
 	errCh := make(chan error, 1)
@@ -434,6 +470,25 @@
 	c.Check(err, Equals, io.ErrUnexpectedEOF)
 }
 
+func (s *sessionSuite) TestSessionLoopConnBrokenExchange(c *C) {
+	nopTrack := NewTracker(s.testlog)
+	errCh := make(chan error, 1)
+	up := make(chan interface{}, 5)
+	down := make(chan interface{}, 5)
+	tp := &testProtocol{up, down}
+	exchanges := make(chan broker.Exchange, 1)
+	exchanges <- &broker.ConnBrokenExchange{"REASON"}
+	sess := &testing.TestBrokerSession{Exchanges: exchanges}
+	go func() {
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
+	}()
+	c.Check(takeNext(down), Equals, "deadline 2ms")
+	c.Check(takeNext(down), DeepEquals, protocol.ConnBrokenMsg{"connbroken", "REASON"})
+	up <- nil // no write error
+	err := <-errCh
+	c.Check(err, DeepEquals, &broker.ErrAbort{"session broken for reason"})
+}
+
 type testTracker struct {
 	SessionTracker
 	interval chan interface{}