Stop processing a response if the Store entry is gone (#806)

HttpStateData::processReply() is usually called synchronously, after
checking the Store entry status, but there are other call chains.

StoreEntry::isAccepting() adds STORE_PENDING check to the ENTRY_ABORTED
check. An accepting entry is required for writing into Store. In theory,
an entry may stop accepting new writes (without being aborted) if
FwdState or another entry co-owner writes an error response due to a
timeout or some other problem that happens while we are waiting for an
I/O callback or some such.

N.B. HTTP and FTP code cannot use StoreEntry::isAccepting() directly
because their network readers may not be the ones writing into Store --
the content may go through the adaptation layer first and that layer
might complete the store entry before the entire peer response is
received. For example, imagine an adaptation service that wants to log
the whole response containing a virus but also replaces that (large)
response with a small error reply.

diff --git a/src/gopher.cc b/src/gopher.cc
index 5127553699ccd45a3668f58d9dfc1c72c26930a9..169b0e1829996ae887cc74dd77ca9bfa721b3d6f 100644 (file)
--- a/src/gopher.cc
+++ b/src/gopher.cc
@@ -740,7 +740,10 @@ gopherReadReply(const Comm::ConnectionPointer &conn, char *buf, size_t len, Comm
 
     assert(buf == gopherState->replybuf);
 
-    if (EBIT_TEST(entry->flags, ENTRY_ABORTED)) {
+    // XXX: Should update delayId, statCounter, etc. before bailing
+    if (!entry->isAccepting()) {
+        debugs(10, 3, "terminating due to bad " << *entry);
+        // TODO: Do not abuse connection for triggering cleanup.
         gopherState->serverConn->close();
         return;
     }
@@ -837,6 +840,13 @@ gopherSendComplete(const Comm::ConnectionPointer &conn, char *, size_t size, Com
         statCounter.server.other.kbytes_out += size;
     }
 
+    if (!entry->isAccepting()) {
+        debugs(10, 3, "terminating due to bad " << *entry);
+        // TODO: Do not abuse connection for triggering cleanup.
+        gopherState->serverConn->close();
+        return;
+    }
+
     if (errflag) {
         ErrorState *err;
         err = new ErrorState(ERR_WRITE_ERROR, Http::scServiceUnavailable, gopherState->fwd->request);

diff --git a/src/http.cc b/src/http.cc
index 691100b078eb5f267fcd17ccbccd510b171bfd5d..017e49247f71a8161cf41ed98edb85ffc09bb063 100644 (file)
--- a/src/http.cc
+++ b/src/http.cc
@@ -1226,6 +1226,11 @@ HttpStateData::processReply()
         Must(!flags.headers_parsed);
     }
 
+    if (EBIT_TEST(entry->flags, ENTRY_ABORTED)) {
+        abortTransaction("store entry aborted while we were waiting for processReply()");
+        return;
+    }
+
     if (!flags.headers_parsed) { // have not parsed headers yet?
         PROF_start(HttpStateData_processReplyHeader);
         processReplyHeader();

diff --git a/src/urn.cc b/src/urn.cc
index 5fb8a727c91088f4e64bd89d6afa4fe252a99937..74453e1cf48b36cff594bb41e77be34d18af63a6 100644 (file)
--- a/src/urn.cc
+++ b/src/urn.cc
@@ -244,6 +244,12 @@ urnHandleReply(void *data, StoreIOBuffer result)
         return;
     }
 
+    if (!e->isAccepting()) {
+        debugs(52, 3, "terminating due to bad " << *e);
+        delete urnState;
+        return;
+    }
+
     /* Update reqofs to point to where in the buffer we'd be */
     urnState->reqofs += result.length;
 

diff --git a/src/whois.cc b/src/whois.cc
index ebc45ff15cb09fb3d876c143500e641c3969a007..a190befa7c7a82197baf401f87c7fedb43f52d6d 100644 (file)
--- a/src/whois.cc
+++ b/src/whois.cc
@@ -120,6 +120,16 @@ WhoisState::readReply(const Comm::ConnectionPointer &conn, char *aBuffer, size_t
     debugs(75, 3, HERE << conn << " read " << aBufferLength << " bytes");
     debugs(75, 5, "{" << aBuffer << "}");
 
+    // TODO: Honor delay pools.
+
+    // XXX: Update statCounter before bailing
+    if (!entry->isAccepting()) {
+        debugs(52, 3, "terminating due to bad " << *entry);
+        // TODO: Do not abuse connection for triggering cleanup.
+        conn->close();
+        return;
+    }
+
     if (flag != Comm::OK) {
         debugs(50, 2, conn << ": read failure: " << xstrerr(xerrno));