Graceful shutdown

daemon/algod/server.go

@@ -240,6 +240,8 @@ func (s *Server) Stop() {
 	// Attempt to log a shutdown event before we exit...
 	s.log.Event(telemetryspec.ApplicationState, telemetryspec.ShutdownEvent)
 
+	s.node.Stop()
+
 	err := server.Shutdown(context.Background())
 	if err != nil {
 		s.log.Error(err)

network/wsNetwork_test.go

@@ -415,6 +415,9 @@ func (nc *nopConn) NextReader() (int, io.Reader, error) {
 func (nc *nopConn) WriteMessage(int, []byte) error {
 	return nil
 }
+func (nc *nopConn) WriteControl(int, []byte, time.Time) error {
+	return nil
+}
 func (nc *nopConn) SetReadLimit(limit int64) {
 }
 func (nc *nopConn) CloseWithoutFlush() error {
@@ -1061,14 +1064,14 @@ func TestWebsocketNetworkManyIdle(t *testing.T) {
 // TODO: test funcion when some message handler is slow?
 
 func TestWebsocketNetwork_updateUrlHost(t *testing.T) {
+	tlog := logging.TestingLog(t)
 	type fields struct {
 		listener               net.Listener
 		server                 http.Server
 		router                 *mux.Router
 		scheme                 string
 		upgrader               websocket.Upgrader
 		config                 config.Local
-		log                    logging.Logger
 		readBuffer             chan IncomingMessage
 		wg                     sync.WaitGroup
 		handlers               Multiplexer
@@ -1099,11 +1102,11 @@ func TestWebsocketNetwork_updateUrlHost(t *testing.T) {
 		originalAddress string
 		host            string
 	}
-	testFields1 := fields{log: logging.NewLogger()}
-	testFields2 := fields{log: logging.NewLogger()}
-	testFields3 := fields{log: logging.NewLogger()}
-	testFields4 := fields{log: logging.NewLogger()}
-	testFields5 := fields{log: logging.NewLogger()}
+	testFields1 := fields{}
+	testFields2 := fields{}
+	testFields3 := fields{}
+	testFields4 := fields{}
+	testFields5 := fields{}
 
 	tests := []struct {
 		name           string
@@ -1164,7 +1167,7 @@ func TestWebsocketNetwork_updateUrlHost(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			wn := &WebsocketNetwork{
-				log: tt.fields.log,
+				log: tlog,
 			}
 			gotNewAddress, err := wn.updateURLHost(tt.args.originalAddress, net.ParseIP(tt.args.host))
 			if (err != nil) != tt.wantErr {

network/wsPeer.go

@@ -65,6 +65,7 @@ type wsPeerWebsocketConn interface {
 	RemoteAddr() net.Addr
 	NextReader() (int, io.Reader, error)
 	WriteMessage(int, []byte) error
+	WriteControl(int, []byte, time.Time) error
 	SetReadLimit(int64)
 	CloseWithoutFlush() error
 }
@@ -258,6 +259,15 @@ func (wp *wsPeer) readLoop() {
 		msg := IncomingMessage{}
 		mtype, reader, err := wp.conn.NextReader()
 		if err != nil {
+			if ce, ok := err.(*websocket.CloseError); ok {
+				switch ce.Code {
+				case websocket.CloseNormalClosure, websocket.CloseGoingAway:
+					// deliberate close, no error
+					return
+				default:
+					// fall through to reportReadErr
+				}
+			}
 			wp.reportReadErr(err)
 			return
 		}
@@ -476,6 +486,7 @@ func (wp *wsPeer) Close() {
 	atomic.StoreInt32(&wp.didSignalClose, 1)
 	if atomic.CompareAndSwapInt32(&wp.didInnerClose, 0, 1) {
 		close(wp.closing)
+		wp.conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(5*time.Second))
 		wp.conn.CloseWithoutFlush()
 	}
 }