flow timeout cleanup and fix

Flow timeout code worked by luck when checking if a flow still needed
reassembly for app layer inspection or logging. It would check for a
part of raw reassembly (smsg list) to determine if detection was
needed. In this case it would also process app layer cleanup,
including logging.

Introduced AppLayerTransactionGetActive which returns the lowest tx_id
in a direction that still needs some work.

FlowForceReassemblyNeedReassmbly now uses it to determine if the
applayer still needs work.

Converted FlowForceReassemblyForHash to use the checking function
FlowForceReassemblyNeedReassmbly as well, so that checking if a flow
needs work is now unified.

src/app-layer-parser.c

@@ -1220,6 +1220,23 @@ error:
     SCReturnInt(-1);
 }
 
+/**
+ *  \brief Get 'active' tx id, meaning the lowest id that still need work.
+ *
+ *  \retval id tx id
+ */
+uint64_t AppLayerTransactionGetActive(Flow *f, uint8_t flags) {
+    AppLayerProto *p = &al_proto_table[f->alproto];
+    uint64_t log_id = ((AppLayerParserStateStore *)f->alparser)->log_id;
+    uint64_t inspect_id = ((AppLayerParserStateStore *)f->alparser)->
+        inspect_id[flags & STREAM_TOSERVER ? 0 : 1];
+    if (p->logger == TRUE) {
+        return (log_id < inspect_id) ? log_id : inspect_id;
+    } else {
+        return inspect_id;
+    }
+}
+
 void AppLayerTransactionUpdateLogId(Flow *f)
 {
     DEBUG_ASSERT_FLOW_LOCKED(f);

src/app-layer-parser.h

@@ -344,6 +344,7 @@ void AppLayerTransactionUpdateInspectId(Flow *f, uint8_t direction);
  */
 uint64_t AppLayerTransactionGetInspectId(Flow *f, uint8_t flags);
 
+uint64_t AppLayerTransactionGetActive(Flow *f, uint8_t flags);
 
 
 void AppLayerSetEOF(Flow *);

src/flow-manager.c

@@ -228,7 +228,8 @@ static int FlowManagerFlowTimedOut(Flow *f, struct timeval *ts) {
     }
 
     int server = 0, client = 0;
-    if (FlowForceReassemblyNeedReassmbly(f, &server, &client) == 1) {
+    if (!(f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE) &&
+            FlowForceReassemblyNeedReassmbly(f, &server, &client) == 1) {
         FlowForceReassemblyForFlowV2(f, server, client);
         return 0;
     }

src/flow-timeout.c

@@ -292,15 +292,16 @@ static inline Packet *FlowForceReassemblyPseudoPacketGet(int direction,
 int FlowForceReassemblyNeedReassmbly(Flow *f, int *server, int *client) {
     TcpSession *ssn;
 
-    /* looks like we have no flows in this queue */
-    if (f == NULL || (f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE)) {
-        return 0;
+    if (f == NULL) {
+        *server = *client = STREAM_HAS_UNPROCESSED_SEGMENTS_NONE;
+        SCReturnInt(0);
     }
 
     /* Get the tcp session for the flow */
     ssn = (TcpSession *)f->protoctx;
     if (ssn == NULL) {
-        return 0;
+        *server = *client = STREAM_HAS_UNPROCESSED_SEGMENTS_NONE;
+        SCReturnInt(0);
     }
 
     *client = StreamNeedsReassembly(ssn, 0);
@@ -317,13 +318,29 @@ int FlowForceReassemblyNeedReassmbly(Flow *f, int *server, int *client) {
             *server = STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION;
     }
 
+    /* if app layer still needs some love, push through */
+    if (f->alproto != ALPROTO_UNKNOWN && f->alstate != NULL &&
+        AppLayerAlprotoSupportsTxs(f->alproto))
+    {
+        uint64_t total_txs = AppLayerGetTxCnt(f->alproto, f->alstate);
+
+        if (AppLayerTransactionGetActive(f, STREAM_TOCLIENT) < total_txs) {
+            if (*server != STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY)
+                *server = STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION;
+        }
+        if (AppLayerTransactionGetActive(f, STREAM_TOSERVER) < total_txs) {
+            if (*client != STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY)
+                *client = STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION;
+        }
+    }
+
     /* nothing to do */
     if (*client == STREAM_HAS_UNPROCESSED_SEGMENTS_NONE &&
         *server == STREAM_HAS_UNPROCESSED_SEGMENTS_NONE) {
-        return 0;
+        SCReturnInt(0);
     }
 
-    return 1;
+    SCReturnInt(1);
 }
 
 /**
@@ -528,8 +545,10 @@ static inline void FlowForceReassemblyForHash(void)
                 continue;
             }
 
+            (void)FlowForceReassemblyNeedReassmbly(f, &server_ok, &client_ok);
+
             /* ah ah!  We have some unattended toserver segments */
-            if ((client_ok = StreamNeedsReassembly(ssn, 0)) == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
+            if (client_ok == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
                 StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);
 
                 ssn->client.last_ack = (ssn->client.seg_list_tail->seq +
@@ -547,7 +566,7 @@ static inline void FlowForceReassemblyForHash(void)
                 }
             }
             /* oh oh!  We have some unattended toclient segments */
-            if ((server_ok = StreamNeedsReassembly(ssn, 1)) == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
+            if (server_ok == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_REASSEMBLY) {
                 StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);
 
                 ssn->server.last_ack = (ssn->server.seg_list_tail->seq +