Cluster abended test 254 (#100)

* Add a test that mongo Server gets their abended reset as necessary.

See https://github.com/go-mgo/mgo/issues/254 and
https://github.com/go-mgo/mgo/pull/255/files

* Include the patch from Issue 255.

This brings in a test which fails without the patch, and passes with the
patch. Still to be tested, manual tcpkill of a socket.

.gitignore

@@ -0,0 +1,2 @@
+_harness
+

cluster.go

@@ -682,6 +682,7 @@ func (cluster *mongoCluster) AcquireSocket(mode Mode, slaveOk bool, syncTimeout
 				time.Sleep(100 * time.Millisecond)
 				continue
 			} else {
+				// We've managed to successfully reconnect to the master, we are no longer abnormaly ended
 				server.Lock()
 				server.abended = false
 				server.Unlock()

export_test.go

@@ -1,6 +1,7 @@
 package mgo
 
 import (
+	"net"
 	"time"
 )
 
@@ -31,3 +32,15 @@ func HackSyncSocketTimeout(newTimeout time.Duration) (restore func()) {
 	syncSocketTimeout = newTimeout
 	return
 }
+
+func (s *Session) Cluster() *mongoCluster {
+	return s.cluster()
+}
+
+func (cluster *mongoCluster) Server(addr string) *mongoServer {
+	tcpaddr, err := net.ResolveTCPAddr("tcp", addr)
+	if err != nil {
+		panic(err)
+	}
+	return cluster.server(addr, tcpaddr)
+}

harness/setup.sh

@@ -30,7 +30,7 @@ start() {
         UP=$(svstat daemons/* | grep ' up ' | grep -v ' [0-3] seconds' | wc -l)
         echo "$UP processes up..."
         if [ x$COUNT = x$UP ]; then
-            echo "Running setup.js with mongo..."
+            echo "Running init.js with mongo..."
             mongo --nodb ../harness/mongojs/init.js
             exit 0
         fi

server_test.go

@@ -0,0 +1,64 @@
+// mgo - MongoDB driver for Go
+//
+// Copyright (c) 2018 Canonical Ltd
+//
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+package mgo_test
+
+import (
+	"time"
+
+	. "gopkg.in/check.v1"
+	"github.com/globalsign/mgo"
+)
+
+func (s *S) TestServerRecoversFromAbend(c *C) {
+	session, err := mgo.Dial("localhost:40001")
+	c.Assert(err, IsNil)
+	defer session.Close()
+	// Peek behind the scenes
+	cluster := session.Cluster()
+	server := cluster.Server("127.0.0.1:40001")
+	sock, abended, err := server.AcquireSocket(100, time.Second)
+	c.Assert(err, IsNil)
+	c.Assert(sock, NotNil)
+	sock.Release()
+	c.Check(abended, Equals, false)
+	// Forcefully abend this socket
+	sock.Close()
+	server.AbendSocket(sock)
+	// Next acquire notices the connection was abnormally ended
+	sock, abended, err = server.AcquireSocket(100, time.Second)
+	c.Assert(err, IsNil)
+	sock.Release()
+	c.Check(abended, Equals, true)
+	// cluster.AcquireSocket should fix the abended problems
+	sock, err = cluster.AcquireSocket(mgo.Primary, false, time.Minute, time.Second, nil, 100)
+	c.Assert(err, IsNil)
+	sock.Release()
+	sock, abended, err = server.AcquireSocket(100, time.Second)
+	c.Assert(err, IsNil)
+	c.Check(abended, Equals, false)
+	sock.Release()
+}