tcp: flow: fixes

engine/tcp_flow.go

@@ -58,11 +58,12 @@ func (f *tcpFlow) feed(l3 L3Info, tcp TCPInfo, payload []byte) io.Verdict {
 
 	if tcp.RST || tcp.FIN {
 		f.closeActiveEntries()
-		f.runMatch(rs, version, rulesetChanged)
+		f.runMatch(rs, version, rulesetChanged, true)
 		f.maybeFinalizeVerdict()
 		return f.lastVerdict
 	}
 
+	propUpdated := false
 	if len(payload) > 0 {
 		dir, rev := f.resolveDirection(tcp)
 		expected := f.dirSeq[dir]
@@ -71,17 +72,18 @@ func (f *tcpFlow) feed(l3 L3Info, tcp TCPInfo, payload []byte) io.Verdict {
 			f.dirBuf[dir] = append(f.dirBuf[dir], payload...)
 			f.dirSeq[dir] = tcp.Seq + uint32(len(payload))
 			if len(f.dirBuf[dir]) <= tcpFlowMaxBuffer {
-				f.feedAnalyzers(rev)
+				propUpdated = f.feedAnalyzers(rev)
 			}
 		}
 	}
 
-	f.runMatch(rs, version, rulesetChanged)
+	f.runMatch(rs, version, rulesetChanged, propUpdated)
 	f.maybeFinalizeVerdict()
 	return f.lastVerdict
 }
 
-func (f *tcpFlow) feedAnalyzers(rev bool) {
+func (f *tcpFlow) feedAnalyzers(rev bool) bool {
+	updated := false
 	buf := f.dirBuf[uint8(tcpDirC2S)]
 	if rev {
 		buf = f.dirBuf[uint8(tcpDirS2C)]
@@ -92,6 +94,7 @@ func (f *tcpFlow) feedAnalyzers(rev bool) {
 		u1 := processPropUpdate(f.info.Props, entry.Name, update)
 		u2 := processPropUpdate(f.info.Props, entry.Name, closeUpdate)
 		if u1 || u2 {
+			updated = true
 			f.logger.TCPStreamPropUpdate(f.info, false)
 		}
 		if done {
@@ -99,10 +102,11 @@ func (f *tcpFlow) feedAnalyzers(rev bool) {
 			f.doneEntries = append(f.doneEntries, entry)
 		}
 	}
+	return updated
 }
 
-func (f *tcpFlow) runMatch(rs ruleset.Ruleset, version uint64, rulesetChanged bool) {
-	if !f.virgin && !rulesetChanged {
+func (f *tcpFlow) runMatch(rs ruleset.Ruleset, version uint64, rulesetChanged bool, propUpdated bool) {
+	if !propUpdated && !f.virgin && !rulesetChanged {
 		return
 	}
 	f.virgin = false