+ testname, expectedBody, response)
+ }
+}
+
+// See #7121
+func TestPutNeedsOnlyOneBuffer(t *testing.T) {
+ defer teardown()
+ KeepVM = MakeTestVolumeManager(1)
+ defer KeepVM.Close()
+
+ defer func(orig *bufferPool) {
+ bufs = orig
+ }(bufs)
+ bufs = newBufferPool(1, BlockSize)
+
+ ok := make(chan struct{})
+ go func() {
+ for i := 0; i < 2; i++ {
+ response := IssueRequest(
+ &RequestTester{
+ method: "PUT",
+ uri: "/" + TestHash,
+ requestBody: TestBlock,
+ })
+ ExpectStatusCode(t,
+ "TestPutNeedsOnlyOneBuffer", http.StatusOK, response)
+ }
+ ok <- struct{}{}
+ }()
+
+ select {
+ case <-ok:
+ case <-time.After(time.Second):
+ t.Fatal("PUT deadlocks with MaxBuffers==1")
+ }
+}
+
+// Invoke the PutBlockHandler a bunch of times to test for bufferpool resource
+// leak.
+func TestPutHandlerNoBufferleak(t *testing.T) {
+ defer teardown()
+
+ // Prepare two test Keep volumes.
+ KeepVM = MakeTestVolumeManager(2)
+ defer KeepVM.Close()
+
+ ok := make(chan bool)
+ go func() {
+ for i := 0; i < theConfig.MaxBuffers+1; i++ {
+ // Unauthenticated request, no server key
+ // => OK (unsigned response)
+ unsignedLocator := "/" + TestHash
+ response := IssueRequest(
+ &RequestTester{
+ method: "PUT",
+ uri: unsignedLocator,
+ requestBody: TestBlock,
+ })
+ ExpectStatusCode(t,
+ "TestPutHandlerBufferleak", http.StatusOK, response)
+ ExpectBody(t,
+ "TestPutHandlerBufferleak",
+ TestHashPutResp, response)
+ }
+ ok <- true
+ }()
+ select {
+ case <-time.After(20 * time.Second):
+ // If the buffer pool leaks, the test goroutine hangs.
+ t.Fatal("test did not finish, assuming pool leaked")
+ case <-ok:
+ }
+}
+
+type notifyingResponseRecorder struct {
+ *httptest.ResponseRecorder
+ closer chan bool
+}
+
+func (r *notifyingResponseRecorder) CloseNotify() <-chan bool {
+ return r.closer
+}
+
+func TestGetHandlerClientDisconnect(t *testing.T) {
+ defer func(was bool) {
+ theConfig.RequireSignatures = was
+ }(theConfig.RequireSignatures)
+ theConfig.RequireSignatures = false
+
+ defer func(orig *bufferPool) {
+ bufs = orig
+ }(bufs)
+ bufs = newBufferPool(1, BlockSize)
+ defer bufs.Put(bufs.Get(BlockSize))
+
+ KeepVM = MakeTestVolumeManager(2)
+ defer KeepVM.Close()
+
+ if err := KeepVM.AllWritable()[0].Put(context.Background(), TestHash, TestBlock); err != nil {
+ t.Error(err)
+ }
+
+ resp := ¬ifyingResponseRecorder{
+ ResponseRecorder: httptest.NewRecorder(),
+ closer: make(chan bool, 1),
+ }
+ if _, ok := http.ResponseWriter(resp).(http.CloseNotifier); !ok {
+ t.Fatal("notifyingResponseRecorder is broken")
+ }
+ // If anyone asks, the client has disconnected.
+ resp.closer <- true
+
+ ok := make(chan struct{})
+ go func() {
+ req, _ := http.NewRequest("GET", fmt.Sprintf("/%s+%d", TestHash, len(TestBlock)), nil)
+ (&LoggingRESTRouter{router: MakeRESTRouter()}).ServeHTTP(resp, req)
+ ok <- struct{}{}
+ }()
+
+ select {
+ case <-time.After(20 * time.Second):
+ t.Fatal("request took >20s, close notifier must be broken")
+ case <-ok:
+ }
+
+ ExpectStatusCode(t, "client disconnect", http.StatusServiceUnavailable, resp.ResponseRecorder)
+ for i, v := range KeepVM.AllWritable() {
+ if calls := v.(*MockVolume).called["GET"]; calls != 0 {
+ t.Errorf("volume %d got %d calls, expected 0", i, calls)
+ }
+ }
+}
+
+// Invoke the GetBlockHandler a bunch of times to test for bufferpool resource
+// leak.
+func TestGetHandlerNoBufferLeak(t *testing.T) {
+ defer teardown()
+
+ // Prepare two test Keep volumes. Our block is stored on the second volume.
+ KeepVM = MakeTestVolumeManager(2)
+ defer KeepVM.Close()
+
+ vols := KeepVM.AllWritable()
+ if err := vols[0].Put(context.Background(), TestHash, TestBlock); err != nil {
+ t.Error(err)
+ }
+
+ ok := make(chan bool)
+ go func() {
+ for i := 0; i < theConfig.MaxBuffers+1; i++ {
+ // Unauthenticated request, unsigned locator
+ // => OK
+ unsignedLocator := "/" + TestHash
+ response := IssueRequest(
+ &RequestTester{
+ method: "GET",
+ uri: unsignedLocator,
+ })
+ ExpectStatusCode(t,
+ "Unauthenticated request, unsigned locator", http.StatusOK, response)
+ ExpectBody(t,
+ "Unauthenticated request, unsigned locator",
+ string(TestBlock),
+ response)
+ }
+ ok <- true
+ }()
+ select {
+ case <-time.After(20 * time.Second):
+ // If the buffer pool leaks, the test goroutine hangs.
+ t.Fatal("test did not finish, assuming pool leaked")
+ case <-ok:
+ }
+}
+
+func TestPutReplicationHeader(t *testing.T) {
+ defer teardown()
+
+ KeepVM = MakeTestVolumeManager(2)
+ defer KeepVM.Close()
+
+ resp := IssueRequest(&RequestTester{
+ method: "PUT",
+ uri: "/" + TestHash,
+ requestBody: TestBlock,
+ })
+ if r := resp.Header().Get("X-Keep-Replicas-Stored"); r != "1" {
+ t.Errorf("Got X-Keep-Replicas-Stored: %q, expected %q", r, "1")
+ }
+}
+
+func TestUntrashHandler(t *testing.T) {
+ defer teardown()
+
+ // Set up Keep volumes
+ KeepVM = MakeTestVolumeManager(2)
+ defer KeepVM.Close()
+ vols := KeepVM.AllWritable()
+ vols[0].Put(context.Background(), TestHash, TestBlock)
+
+ theConfig.systemAuthToken = "DATA MANAGER TOKEN"
+
+ // unauthenticatedReq => UnauthorizedError
+ unauthenticatedReq := &RequestTester{
+ method: "PUT",
+ uri: "/untrash/" + TestHash,