10467: Add "client disconnect" test.

diff --git a/services/keepstore/s3_volume_test.go b/services/keepstore/s3_volume_test.go
index 6389d503dfc5ccee32471d308bf81ee64b012135..702e5539dfb0f060360120bc4312ce391b523b0c 100644 (file)
--- a/services/keepstore/s3_volume_test.go
+++ b/services/keepstore/s3_volume_test.go
@@ -7,6 +7,8 @@ import (
        "encoding/json"
        "fmt"
        "io/ioutil"
+       "net/http"
+       "net/http/httptest"
        "os"
        "time"
 
@@ -112,6 +114,73 @@ func (s *StubbedS3Suite) TestStats(c *check.C) {
        c.Check(stats(), check.Matches, `.*"InBytes":6,.*`)
 }
 
+type blockingHandler struct {
+       requested chan *http.Request
+       unblock   chan struct{}
+}
+
+func (h *blockingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+       if h.requested != nil {
+               h.requested <- r
+       }
+       if h.unblock != nil {
+               <-h.unblock
+       }
+       http.Error(w, "nothing here", http.StatusNotFound)
+}
+
+func (s *StubbedS3Suite) TestClientDisconnect(c *check.C) {
+       loc := "acbd18db4cc2f85cedef654fccc4a4d8"
+       buf := make([]byte, 3)
+
+       handler := &blockingHandler{}
+       srv := httptest.NewServer(handler)
+       defer srv.Close()
+
+       v := s.newTestableVolume(c, 5*time.Minute, false, 2)
+       vol := *v.S3Volume
+       vol.Endpoint = srv.URL
+       v = &TestableS3Volume{S3Volume: &vol}
+       v.Start()
+
+       ctx, cancel := context.WithCancel(context.Background())
+
+       handler.requested = make(chan *http.Request)
+       handler.unblock = make(chan struct{})
+       defer close(handler.unblock)
+
+       var n int
+       var err error
+       doneGet := make(chan struct{})
+       go func() {
+               n, err = v.Get(ctx, loc, buf)
+               close(doneGet)
+       }()
+
+       timeout := time.After(10 * time.Second)
+
+       // Wait for the stub server to receive a request, meaning
+       // Get() is waiting for an s3 operation.
+       select {
+       case <-timeout:
+               c.Fatal("timed out waiting for Get to call our handler")
+       case <-doneGet:
+               c.Fatal("Get finished without calling our handler!")
+       case <-handler.requested:
+       }
+
+       cancel()
+
+       select {
+       case <-timeout:
+               c.Fatal("timed out")
+       case <-doneGet:
+               c.Check(n, check.Equals, 0)
+               c.Check(err, check.NotNil)
+               c.Check(err, check.Equals, context.Canceled)
+       }
+}
+
 func (s *StubbedS3Suite) TestBackendStates(c *check.C) {
        defer func(tl, bs arvados.Duration) {
                theConfig.TrashLifetime = tl
@@ -320,18 +389,9 @@ func (s *StubbedS3Suite) newTestableVolume(c *check.C, raceWindow time.Duration,
        srv, err := s3test.NewServer(&s3test.Config{Clock: clock})
        c.Assert(err, check.IsNil)
 
-       tmp, err := ioutil.TempFile("", "keepstore")
-       c.Assert(err, check.IsNil)
-       defer os.Remove(tmp.Name())
-       _, err = tmp.Write([]byte("xxx\n"))
-       c.Assert(err, check.IsNil)
-       c.Assert(tmp.Close(), check.IsNil)
-
        v := &TestableS3Volume{
                S3Volume: &S3Volume{
                        Bucket:             TestBucketName,
-                       AccessKeyFile:      tmp.Name(),
-                       SecretKeyFile:      tmp.Name(),
                        Endpoint:           srv.URL(),
                        Region:             "test-region-1",
                        LocationConstraint: true,
@@ -341,15 +401,31 @@ func (s *StubbedS3Suite) newTestableVolume(c *check.C, raceWindow time.Duration,
                        ReadOnly:           readonly,
                        IndexPageSize:      1000,
                },
+               c:           c,
                server:      srv,
                serverClock: clock,
        }
-       c.Assert(v.Start(), check.IsNil)
+       v.Start()
        err = v.bucket.PutBucket(s3.ACL("private"))
        c.Assert(err, check.IsNil)
        return v
 }
 
+func (v *TestableS3Volume) Start() error {
+       tmp, err := ioutil.TempFile("", "keepstore")
+       v.c.Assert(err, check.IsNil)
+       defer os.Remove(tmp.Name())
+       _, err = tmp.Write([]byte("xxx\n"))
+       v.c.Assert(err, check.IsNil)
+       v.c.Assert(tmp.Close(), check.IsNil)
+
+       v.S3Volume.AccessKeyFile = tmp.Name()
+       v.S3Volume.SecretKeyFile = tmp.Name()
+
+       v.c.Assert(v.S3Volume.Start(), check.IsNil)
+       return nil
+}
+
 // PutRaw skips the ContentMD5 test
 func (v *TestableS3Volume) PutRaw(loc string, block []byte) {
        err := v.bucket.Put(loc, block, "application/octet-stream", s3ACL, s3.Options{})