Resolves #99: Clean internal state of DocTransaction on tr->onError() (…

…#110)

By not cleaning DocTransaction state which internally depends on FDB
transaction state, concurrent actors during onError() are causing
segfault.

src/ExtCmd.actor.cpp

@@ -1224,7 +1224,7 @@ struct ListCollectionsCmd {
 					break;
 				}
 				if (e.code() != error_code_actor_cancelled) {
-					Void _ = wait(dtr->tr->onError(e));
+					Void _ = wait(dtr->onError(e));
 				}
 			}
 		}
@@ -1304,7 +1304,7 @@ struct ListIndexesCmd {
 				return reply;
 			} catch (Error& e) {
 				if (e.code() != error_code_actor_cancelled) {
-					Void _ = wait(dtr->tr->onError(e));
+					Void _ = wait(dtr->onError(e));
 				}
 			}
 		}

src/QLContext.actor.cpp

@@ -52,6 +52,18 @@ ACTOR Future<Void> DocumentDeferred::commitChanges(Reference<DocTransaction> tr,
 	return Void();
 }
 
+/**
+ * Reset all the state in DocTransaction and call onError on transaction. Its important to
+ * cancel the pending actors before calling `tr->onError()`. onError on FDB transaction
+ * resets the transaction, concurrent actors using this Transaction will have inconsistent
+ * behaviour, if not cancelled.
+ */
+Future<Void> DocTransaction::onError(Error const& e) {
+	cancel_ongoing_index_reads();
+	infos.clear();
+	return tr->onError(e);
+}
+
 void DocTransaction::cancel_ongoing_index_reads() {
 	for (auto it : infos) {
 		for (auto actor : it.second->index_update_actors)

src/QLContext.h

@@ -85,6 +85,8 @@ struct DocTransaction : ReferenceCounted<DocTransaction> {
 	}
 	static Future<Void> commitChanges(Reference<DocTransaction> const& self, std::string const& docPrefix);
 
+	Future<Void> onError(Error const& e);
+
 	// If you are about to call this function, think very carefully about why you are doing that. It is not safe to call
 	// in general. Look at the comments in doNonIsolatedRW() for more on what it's for.
 	void cancel_ongoing_index_reads();

src/QLPlan.actor.cpp

@@ -857,7 +857,7 @@ ACTOR static Future<Void> doNonIsolatedRW(PlanCheckpoint* outerCheckpoint,
 					bufferedDocs.pop_front();
 				}
 			} catch (Error& e) {
-				Void _ = wait(dtr->tr->onError(e));
+				Void _ = wait(dtr->onError(e));
 				finished = false;
 			}
 
@@ -957,7 +957,7 @@ ACTOR static Future<Void> doRetry(Reference<Plan> subPlan,
 					throw;
 				if (e.code() == error_code_end_of_stream)
 					throw;
-				Void _ = wait(tr->tr->onError(e));
+				Void _ = wait(tr->onError(e));
 				tr = self->newTransaction(); // FIXME: keep dtr->tr if this is a retry
 			}
 		}