Expire pending requests instead of keeping them for ever

src/hashtable.c

@@ -61,6 +61,7 @@ hashtable *hashtable_create(int nbucket) {
    if (!ht) return NULL;
    ht->free = NULL;
    ht->nbucket = nbucket;
+   ht->size = 0;
    ht->buckets = calloc(nbucket, sizeof(entry*));
    if (!ht->buckets) {
       free(ht);
@@ -102,10 +103,16 @@ void *hashtable_add(hashtable *ht, char *key, void *value) {
         current = current->next;
     }
     entry *e = malloc(sizeof(*e));
-    e->value = value;
+    if (!e) return NULL;
     e->key = strdup(key);
+    if (!e->key) {
+        free(e);
+        return NULL;
+    }
+    e->value = value;
     e->next = ht->buckets[bucket];
     ht->buckets[bucket] = e;
+    ht->size++;
     return NULL;
 }
 
@@ -141,6 +148,7 @@ int hashtable_del(hashtable *ht, char *key) {
             free(current->key);
             ht->free ? ht->free(current->value):free(current->value);
             free(current);
+            ht->size--;
             return 1;
         }
         prev = current;
@@ -149,6 +157,36 @@ int hashtable_del(hashtable *ht, char *key) {
     return 0;
 }
 
+int hashtable_sweep(hashtable *ht, int (*expired)(void *value, void *arg), void *arg) {
+    int i, removed = 0;
+    entry *current, *next, *prev;
+
+    if (!ht || !expired) return 0;
+    for (i = 0; i < ht->nbucket; i++) {
+        prev = NULL;
+        current = ht->buckets[i];
+        while (current) {
+            next = current->next;
+            if (expired(current->value, arg)) {
+                if (prev) {
+                    prev->next = next;
+                } else {
+                    ht->buckets[i] = next;
+                }
+                free(current->key);
+                ht->free ? ht->free(current->value) : free(current->value);
+                free(current);
+                ht->size--;
+                removed++;
+            } else {
+                prev = current;
+            }
+            current = next;
+        }
+    }
+    return removed;
+}
+
 void **hashtable_values(hashtable *ht, int *cnt) {
     int i, cap = 0;
     entry *e;

src/hashtable.h

@@ -20,6 +20,7 @@ typedef struct entry {
 
 typedef struct hashtable {
     int nbucket;
+    int size;
     entry **buckets;
     void (*free)(void *);
 } hashtable;
@@ -30,4 +31,6 @@ void *hashtable_get(hashtable *ht, char *key);
 void *hashtable_add(hashtable *ht, char *key, void *value);
 int hashtable_del(hashtable *ht, char *key);
 void **hashtable_values(hashtable *ht, int *cnt);
+/* Removes every entry the predicate accepts and returns how many went. */
+int hashtable_sweep(hashtable *ht, int (*expired)(void *value, void *arg), void *arg);
 #endif

src/packet.c

@@ -117,6 +117,50 @@ int process_udp_packet(const struct timeval tv,
     return 0;
 }
 
+/* A response later than this is past the last latency bucket, so a request
+ * still waiting after it is treated as lost rather than kept for ever. */
+#define REQUEST_TIMEOUT_US 60000000
+/* Capture time between sweeps of the pending table. */
+#define EXPIRE_INTERVAL_US 1000000
+/* Firm ceiling for the case where requests arrive faster than they expire. */
+#define MAX_PENDING_REQUESTS 100000
+
+struct request_expiry {
+    struct timeval now;
+    int64_t timeout_us;
+};
+
+static int64_t tv_delta_us(struct timeval later, struct timeval earlier) {
+    return (int64_t)(later.tv_sec - earlier.tv_sec) * 1000000
+        + (later.tv_usec - earlier.tv_usec);
+}
+
+static int request_expired(void *value, void *arg) {
+    const struct request *req = (const struct request *)value;
+    const struct request_expiry *expiry = (const struct request_expiry *)arg;
+
+    return tv_delta_us(expiry->now, req->tv) > expiry->timeout_us;
+}
+
+int requests_expire(struct hashtable *requests, struct timeval now, int64_t timeout_us) {
+    struct request_expiry expiry;
+
+    expiry.now = now;
+    expiry.timeout_us = timeout_us;
+    return hashtable_sweep(requests, request_expired, &expiry);
+}
+
+static void expire_stale_requests(struct sniffer *sniffer, struct timeval now) {
+    if (!sniffer->expire_primed) {
+        sniffer->expire_primed = 1;
+        sniffer->last_expire = now;
+        return;
+    }
+    if (tv_delta_us(now, sniffer->last_expire) < EXPIRE_INTERVAL_US) return;
+    sniffer->last_expire = now;
+    requests_expire(sniffer->requests, now, REQUEST_TIMEOUT_US);
+}
+
 static int packet_direction(struct sniffer *sniffer, struct user_packet *upacket) {
     char key[32];
     int direction = -1;
@@ -172,6 +216,8 @@ static void process_request_packet(struct sniffer *sniffer, struct user_packet *
             upacket->ip_src.s_addr, upacket->port_src,
             upacket->ip_dst.s_addr, upacket->port_dst);
     if (hashtable_get(sniffer->requests, key)) return;
+    /* Drop the request rather than grow the table without limit. */
+    if (sniffer->requests->size >= MAX_PENDING_REQUESTS) return;
 
     req = malloc(sizeof(*req));
     if (!req) return;
@@ -205,8 +251,7 @@ static void process_response_packet(struct sniffer *sniffer, struct user_packet
             upacket->ip_dst.s_addr, upacket->port_dst,
             upacket->ip_src.s_addr, upacket->port_src);
     if ((req = hashtable_get(sniffer->requests, key)) != NULL) {
-        delta = (upacket->tv.tv_sec - req->tv.tv_sec) * 1000000
-            + (upacket->tv.tv_usec - req->tv.tv_usec);
+        delta = tv_delta_us(upacket->tv, req->tv);
         snprintf(target, sizeof(target), "%u:%d", upacket->ip_src.s_addr, upacket->port_src);
         sniffer_stats_lock(sniffer);
         if ((stats = hashtable_get(sniffer->syn_tab, target)) != NULL) {
@@ -295,6 +340,7 @@ void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket) {
         print_user_packet(sniffer, upacket);
         return;
     }
+    expire_stale_requests(sniffer, upacket->tv);
     if (upacket->payload_size == 0) {
         if ((upacket->flags & syn_mask) != 0) {
             src = (upacket->flags & ack_mask) != 0; 

src/packet.h

@@ -62,5 +62,8 @@ int process_udp_packet(const struct timeval tv,
         int wirelen,
         struct user_packet *packet);
 
+/* Drops pending requests older than timeout_us and returns how many went. */
+int requests_expire(struct hashtable *requests, struct timeval now, int64_t timeout_us);
+
 void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket); 
 #endif

src/sniffer.h

@@ -14,6 +14,7 @@
 
 #include <pcap.h>
 #include <pthread.h>
+#include <sys/time.h>
 #include <lua.h>
 #include "tcpkit.h"
 #include "stats.h"
@@ -34,6 +35,8 @@ struct sniffer {
     int lock_ready;
     struct hashtable *syn_tab;
     struct hashtable *requests;
+    struct timeval last_expire;
+    int expire_primed;
     lua_State *lua_state;
     struct bpf_program *bpf;
 };

tests/e2e/run.sh

@@ -127,6 +127,13 @@ expect_lines "each request stays on one line" "$out" 3
 expect_contains "a truncated resp array is summarised" "$out" '*2..$3..GET'
 expect_contains "a payload with no crlf is summarised" "$out" "GETNOCRLFATALL"
 
+echo "== e2e: stale requests"
+replay stale-request.pcap -p redis
+expect_status "a stale request exits cleanly" "$status" 0
+expect_lines "only the answered request is reported" "$out" 1
+expect_contains "the healthy exchange is still reported" "$out" "GET fast"
+expect_not_contains "the abandoned request is not held for ever" "$out" "GET slow"
+
 echo "== e2e: capture filter"
 replay redis-session.pcap -p raw udp
 expect_lines "a udp filter keeps only the dns packet" "$out" 1

tests/fixtures/gen_fixtures.py

@@ -184,6 +184,21 @@ def malformed_payload():
     ]
 
 
+def stale_request():
+    """A request whose response arrives long after any latency bucket, followed
+    by a healthy exchange, so the sweep can be told apart from a broken path."""
+    return [
+        full(0.000000, to_server(b"", 1000, 0, SYN)),
+        full(0.000100, to_client(b"", 5000, 1001, SYN | ACK)),
+        # Answered 99 seconds later, well past the request timeout.
+        full(1.000000, to_server(resp("GET", "slow"), 1001, 5001, PSH | ACK)),
+        full(100.000000, to_client(b"$1\r\nb\r\n", 5001, 1024, PSH | ACK)),
+        # A normal exchange afterwards still has to be reported.
+        full(101.000000, to_server(resp("GET", "fast"), 1024, 5008, PSH | ACK)),
+        full(101.000500, to_client(b"$1\r\nc\r\n", 5008, 1047, PSH | ACK)),
+    ]
+
+
 def stress(connections=4000):
     """Many short-lived connections, to keep the capture thread busy while the
     stats endpoint is polled. Not a committed fixture: it is generated on demand
@@ -208,3 +223,4 @@ def stress(connections=4000):
         write_pcap("redis-session.pcap", redis_session())
         write_pcap("truncated.pcap", truncated())
         write_pcap("malformed-payload.pcap", malformed_payload())
+        write_pcap("stale-request.pcap", stale_request())

tests/unit/test_hashtable.c

@@ -99,6 +99,95 @@ static void test_values_lists_every_entry(void) {
     hashtable_destroy(ht);
 }
 
+static int always(void *value, void *arg) {
+    (void)value;
+    (void)arg;
+    return 1;
+}
+
+static int value_is(void *value, void *arg) {
+    return strcmp((char *)value, (char *)arg) == 0;
+}
+
+static void test_size_tracks_the_entries(void) {
+    hashtable *ht = hashtable_create(8);
+    char *rejected;
+
+    TK_EQ_INT(ht->size, 0);
+    hashtable_add(ht, "a", dup_value("1"));
+    hashtable_add(ht, "b", dup_value("2"));
+    TK_EQ_INT(ht->size, 2);
+
+    /* A duplicate key adds no entry and hands back the value already there;
+     * the rejected one still belongs to the caller. */
+    rejected = dup_value("3");
+    TK_EQ_STR((char *)hashtable_add(ht, "a", rejected), "1");
+    free(rejected);
+    TK_EQ_INT(ht->size, 2);
+
+    hashtable_del(ht, "a");
+    TK_EQ_INT(ht->size, 1);
+    hashtable_del(ht, "missing");
+    TK_EQ_INT(ht->size, 1);
+
+    hashtable_destroy(ht);
+}
+
+static void test_sweep_removes_only_what_matches(void) {
+    hashtable *ht = hashtable_create(4);
+
+    hashtable_add(ht, "a", dup_value("keep"));
+    hashtable_add(ht, "b", dup_value("drop"));
+    hashtable_add(ht, "c", dup_value("drop"));
+    hashtable_add(ht, "d", dup_value("keep"));
+
+    TK_EQ_INT(hashtable_sweep(ht, value_is, (void *)"drop"), 2);
+    TK_EQ_INT(ht->size, 2);
+    TK_EQ_STR((char *)hashtable_get(ht, "a"), "keep");
+    TK_EQ_STR((char *)hashtable_get(ht, "d"), "keep");
+    TK_CHECK(hashtable_get(ht, "b") == NULL, "swept key must be gone");
+    TK_CHECK(hashtable_get(ht, "c") == NULL, "swept key must be gone");
+
+    hashtable_destroy(ht);
+}
+
+static void test_sweep_handles_a_whole_bucket_chain(void) {
+    hashtable *ht = hashtable_create(1);
+    char key[16];
+    int i;
+
+    /* One bucket, so every entry is chained: head, middle and tail all go. */
+    for (i = 0; i < 16; i++) {
+        snprintf(key, sizeof(key), "key-%d", i);
+        hashtable_add(ht, key, dup_value("drop"));
+    }
+    hashtable_add(ht, "survivor", dup_value("keep"));
+
+    TK_EQ_INT(hashtable_sweep(ht, value_is, (void *)"drop"), 16);
+    TK_EQ_INT(ht->size, 1);
+    TK_EQ_STR((char *)hashtable_get(ht, "survivor"), "keep");
+
+    TK_EQ_INT(hashtable_sweep(ht, always, NULL), 1);
+    TK_EQ_INT(ht->size, 0);
+    TK_CHECK(hashtable_get(ht, "survivor") == NULL, "the table must be empty");
+
+    hashtable_destroy(ht);
+}
+
+static void test_sweep_uses_the_custom_free(void) {
+    hashtable *ht = hashtable_create(8);
+
+    freed = 0;
+    ht->free = counting_free;
+    hashtable_add(ht, "a", dup_value("1"));
+    hashtable_add(ht, "b", dup_value("2"));
+
+    TK_EQ_INT(hashtable_sweep(ht, always, NULL), 2);
+    TK_EQ_INT(freed, 2);
+
+    hashtable_destroy(ht);
+}
+
 static void test_destroy_uses_the_custom_free(void) {
     hashtable *ht = hashtable_create(8);
 
@@ -117,6 +206,10 @@ int main(void) {
     TK_RUN(test_del_removes_the_entry);
     TK_RUN(test_collisions_keep_every_entry);
     TK_RUN(test_values_lists_every_entry);
+    TK_RUN(test_size_tracks_the_entries);
+    TK_RUN(test_sweep_removes_only_what_matches);
+    TK_RUN(test_sweep_handles_a_whole_bucket_chain);
+    TK_RUN(test_sweep_uses_the_custom_free);
     TK_RUN(test_destroy_uses_the_custom_free);
     return tk_report("hashtable");
 }