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 6c22c58152fecbaffc5a4d7ad0771e46649afbc1..f1cfb2f77aad74eb77a230ccc33a31f6c4345edd 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 2fb12df6a248dc53a1409ccf3ab12875c24ff38e..f4e033f0760be532297e8436a363f75dc8774030 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 9d65321c4e891c88826e534fc8f7f558b7e873f0..29c121054c2a2a307934051c4d440dbff0e394bc 100644 (file)
--- a/src/urn.cc
+++ b/src/urn.cc
@@ -261,6 +261,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 c12b00e8fcaeee49ed9446f79a06e7ecdc5c5aef..7b3a7cb9a5e978c916ecf93029f28f9bb353e768 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));