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 f33ef5d6dbf5f1aa3ea25ab1576b505ee4fb3e49..4f7a293778ef5e31ee6e065d5bd7753cc0f6670f 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) {
         const auto err = new ErrorState(ERR_WRITE_ERROR, Http::scServiceUnavailable, gopherState->fwd->request, gopherState->fwd->al);
         err->xerrno = xerrno;

diff --git a/src/http.cc b/src/http.cc
index 4735d58d56a92f7773e5c7039134c9b25fe7724b..0dd32501a94578b7f12be9682975a6680bad8e2e 100644 (file)
--- a/src/http.cc
+++ b/src/http.cc
@@ -1314,6 +1314,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 72ab801a9069c265238470b8903b73079ab139e7..83d7ea333db474069bd8a7c69085204eb59fd698 100644 (file)
--- a/src/urn.cc
+++ b/src/urn.cc
@@ -257,6 +257,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 a1a3beb7b1eb59961e733c4a60ccdbfea20db432..f85df20493ff0561c1c0aed5d0174241946388d3 100644 (file)
--- a/src/whois.cc
+++ b/src/whois.cc
@@ -118,6 +118,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));