feat(server): add adaptive tiered message queues for slow clients

html5-client/src/css/jittertrap.css

@@ -442,4 +442,22 @@ footer a {
   padding-right: 1rem;
   border-right: 1px solid #dee2e6;
   margin-right: 1rem !important;
+}
+
+/* Style for disabled interval options when server degrades resolution */
+#chopts_chartPeriod option:disabled {
+  color: #adb5bd;
+  font-style: italic;
+}
+
+/* Resolution indicator shown when degraded from full resolution */
+#resolution-indicator {
+  font-size: 0.75rem;
+  color: #856404;
+  background-color: #fff3cd;
+  border: 1px solid #ffc107;
+  border-radius: 0.25rem;
+  padding: 0.15rem 0.4rem;
+  margin-left: 0.5rem;
+  display: none;
 }

html5-client/src/html/panel-chartparams.part.html

@@ -22,6 +22,7 @@
                 <span class="input-group-text">ms</span>
               </div>
             </div>
+            <span id="resolution-indicator" title="Connection speed limited - some intervals unavailable"></span>
           </div>
 
           <div class="d-flex" id="actionButtons">

html5-client/src/js/jittertrap-websocket.js

@@ -187,6 +187,63 @@
     }
   };
 
+  const handleMsgError = function(params) {
+    console.error("Server error:", params.code, "-", params.text);
+
+    const errorMsgElement = $("#error-msg");
+    const errorModalElement = $("#error-modal");
+
+    let userMessage = params.text || "Unknown server error";
+
+    /* Add specific guidance for known error codes */
+    if (params.code === "max_connections") {
+      userMessage += "<br><br><strong>Tip:</strong> Close other JitterTrap tabs or windows and refresh this page.";
+    } else if (params.code === "too_slow") {
+      userMessage += "<br><br><strong>Tip:</strong> Your connection may be too slow. Try using a faster network or reducing the sample rate.";
+    }
+
+    errorMsgElement.html("<p>" + userMessage + "</p>");
+    errorModalElement.modal('show');
+  };
+
+  /**
+   * Handle resolution message from server - disables interval options that
+   * are faster than the current minimum supported interval.
+   *
+   * The server sends this when the client's connection speed requires
+   * degrading to a slower sample rate tier.
+   */
+  const handleMsgResolution = function(params) {
+    const minIntervalMs = params.min_interval_ms;
+    console.log("Resolution update: min_interval_ms =", minIntervalMs);
+
+    /* Disable interval options faster than current capability */
+    $("#chopts_chartPeriod option").each(function() {
+      const optionVal = parseInt($(this).val(), 10);
+      /* Option value is in ms, same as minIntervalMs */
+      $(this).prop('disabled', optionVal < minIntervalMs);
+    });
+
+    /* If currently selected interval is now disabled, switch to minimum available */
+    const selectedVal = parseInt($("#chopts_chartPeriod").val(), 10);
+    if (selectedVal < minIntervalMs) {
+      console.log("Switching from disabled interval", selectedVal, "to", minIntervalMs);
+      $("#chopts_chartPeriod").val(minIntervalMs).trigger('change');
+    }
+
+    /* Update the resolution indicator if present */
+    const resolutionIndicator = $("#resolution-indicator");
+    if (resolutionIndicator.length) {
+      if (minIntervalMs > 5) {
+        /* Show indicator when degraded from full resolution */
+        resolutionIndicator.text(minIntervalMs + "ms min").show();
+      } else {
+        /* Hide when at full resolution */
+        resolutionIndicator.hide();
+      }
+    }
+  };
+
 
   /**
    * Websocket Sending Functions
@@ -249,6 +306,8 @@
     pcap_config: handleMsgPcapConfig,
     pcap_status: handleMsgPcapStatus,
     pcap_ready: handleMsgPcapReady,
+    error: handleMsgError,
+    resolution: handleMsgResolution,
   };
 
   Object.freeze(messageHandlers); // Prevent modification of messageHandlers

server/Makefile

@@ -30,7 +30,11 @@ SOURCES = \
  server-main.c \
  proto.c \
  proto-jittertrap.c \
- mq_msg_ws.c \
+ mq_msg_ws_1.c \
+ mq_msg_ws_2.c \
+ mq_msg_ws_3.c \
+ mq_msg_ws_4.c \
+ mq_msg_ws_5.c \
  mq_msg_stats.c \
  mq_msg_tt.c \
  jt_server_message_handler.c \
@@ -51,7 +55,12 @@ HEADERS = \
  proto.h \
  proto-jittertrap.h \
  mq_generic.h \
- mq_msg_ws.h \
+ mq_ws_tiered.h \
+ mq_msg_ws_1.h \
+ mq_msg_ws_2.h \
+ mq_msg_ws_3.h \
+ mq_msg_ws_4.h \
+ mq_msg_ws_5.h \
  mq_msg_stats.h \
  mq_msg_tt.h \
  jt_server_message_handler.h \
@@ -71,7 +80,11 @@ OBJECTS += compute_thread.o
 OBJECTS += server-main.o
 OBJECTS += proto.o
 OBJECTS += proto-jittertrap.o
-OBJECTS += mq_msg_ws.o
+OBJECTS += mq_msg_ws_1.o
+OBJECTS += mq_msg_ws_2.o
+OBJECTS += mq_msg_ws_3.o
+OBJECTS += mq_msg_ws_4.o
+OBJECTS += mq_msg_ws_5.o
 OBJECTS += mq_msg_stats.o
 OBJECTS += mq_msg_tt.o
 OBJECTS += jt_server_message_handler.o
@@ -143,12 +156,14 @@ indent:
 
 
 MQ_DEPENDS = mq_generic.c mq_generic.h
-MQ_SOURCES = mq_msg_ws.c mq_msg_stats.c
-MQ_HEADERS = mq_msg_ws.h mq_msg_stats.h
+MQ_WS_TIERED_SOURCES = mq_msg_ws_1.c mq_msg_ws_2.c mq_msg_ws_3.c mq_msg_ws_4.c mq_msg_ws_5.c
+MQ_WS_TIERED_HEADERS = mq_ws_tiered.h mq_msg_ws_1.h mq_msg_ws_2.h mq_msg_ws_3.h mq_msg_ws_4.h mq_msg_ws_5.h
+MQ_SOURCES = $(MQ_WS_TIERED_SOURCES) mq_msg_stats.c
+MQ_HEADERS = $(MQ_WS_TIERED_HEADERS) mq_msg_stats.h
 MQ_TEST_SOURCES = test_mq.c $(MQ_SOURCES)
 MQ_MT_TEST_SOURCES = test_mq_mt.c $(MQ_SOURCES)
 MQ_MULTI_TEST_SOURCES = test_multi_mq.c $(MQ_SOURCES)
-MQ_TEST_HEADERS = mq_msg_ws.h mq_generic.h
+MQ_TEST_HEADERS = $(MQ_WS_TIERED_HEADERS) mq_generic.h
 
 test-mq: $(MQ_TEST_SOURCES) $(MQ_TEST_HEADERS) $(MQ_DEPENDS) $(MQ_HEADERS)
 	$(CC) -o test-mq $(MQ_TEST_SOURCES) $(CFLAGS) -O0 $(DEFINES)
@@ -168,12 +183,22 @@ test-pcap: test_pcap_buffer.c pcap_buffer.o
 .PHONY: test
 test: test-mq test-mq-mt test-multi-mq test-slist test-pcap
 	./test-mq >/dev/null
-	./test-mq-mt >/dev/null
+	./test-mq-mt
 	./test-multi-mq >/dev/null
 	./test-slist
 	./test-pcap
 	@echo -e "Test OK\n"
 
+# Performance benchmark target - builds with optimization and SPEED_TEST
+# Uses high stale threshold (1M) so consumers don't get marked stale during benchmark
+bench-mq-mt: $(MQ_MT_TEST_SOURCES) $(MQ_TEST_HEADERS) $(MQ_HEADERS) $(MQ_DEPENDS)
+	$(CC) -o bench-mq-mt $(MQ_MT_TEST_SOURCES) $(CFLAGS) -O2 -DSPEED_TEST -DMAX_DROPPED_BEFORE_STALE=1000000 $(DEFINES)
+
+.PHONY: benchmark
+benchmark: bench-mq-mt
+	@echo "Running message queue benchmark (1M iterations, no sleeps)..."
+	./bench-mq-mt
+
 TOPTALK_TEST_SOURCES = test-toptalk.c timeywimey.c
 
 test-toptalk: $(TOPTALK_LIB) $(TOPTALK_TEST_SOURCES)
@@ -182,5 +207,5 @@ test-toptalk: $(TOPTALK_LIB) $(TOPTALK_TEST_SOURCES)
 .PHONY: clean
 clean:
 	rm $(PROG) *.o || true
-	rm test-mq test-mq-mt test-multi-mq test-pcap || true
+	rm test-mq test-mq-mt test-multi-mq test-pcap bench-mq-mt || true
 	rm *.gcno *.gcov *.gcda || true

server/jt_server_message_handler.c

@@ -25,7 +25,7 @@
 #include "netem.h"
 
 #include "mq_msg_stats.h"
-#include "mq_msg_ws.h"
+#include "mq_ws_tiered.h"
 #include "mq_msg_tt.h"
 
 #include "jt_message_types.h"
@@ -45,7 +45,7 @@ enum {
 	JT_STATE_PAUSED
 };
 
-int g_jt_state = JT_STATE_STARTING;
+int g_jt_state = JT_STATE_STOPPED;  /* Start stopped, init on first client connect */
 char g_selected_iface[MAX_IFACE_LEN];
 unsigned long stats_consumer_id;
 unsigned long tt_consumer_id;
@@ -137,13 +137,82 @@ static void mq_stats_msg_to_jt_msg_stats(struct mq_stats_msg *mq_s,
 	msg_s->interval_ns = mq_s->interval_ns;
 }
 
-inline static int message_producer(struct mq_ws_msg *m, void *data)
+/* Tier-specific message producers */
+inline static int message_producer_1(struct mq_ws_1_msg *m, void *data)
 {
 	char *s = (char *)data;
 	snprintf(m->m, MAX_JSON_MSG_LEN, "%s", s);
 	return 0;
 }
 
+inline static int message_producer_2(struct mq_ws_2_msg *m, void *data)
+{
+	char *s = (char *)data;
+	snprintf(m->m, MAX_JSON_MSG_LEN, "%s", s);
+	return 0;
+}
+
+inline static int message_producer_3(struct mq_ws_3_msg *m, void *data)
+{
+	char *s = (char *)data;
+	snprintf(m->m, MAX_JSON_MSG_LEN, "%s", s);
+	return 0;
+}
+
+inline static int message_producer_4(struct mq_ws_4_msg *m, void *data)
+{
+	char *s = (char *)data;
+	snprintf(m->m, MAX_JSON_MSG_LEN, "%s", s);
+	return 0;
+}
+
+inline static int message_producer_5(struct mq_ws_5_msg *m, void *data)
+{
+	char *s = (char *)data;
+	snprintf(m->m, MAX_JSON_MSG_LEN, "%s", s);
+	return 0;
+}
+
+/* Route message to appropriate tier based on interval */
+static int jt_srv_send_tiered(int msg_type, void *msg_data, uint64_t interval_ns)
+{
+	char *tmpstr;
+	int cb_err, err = 0;
+	int tier;
+
+	/* convert from jt_msg_* to string */
+	err = jt_messages[msg_type].to_json_string(msg_data, &tmpstr);
+	if (err) {
+		return -1;
+	}
+
+	tier = mq_ws_interval_to_tier(interval_ns);
+
+	/* write the json string to the appropriate tier queue */
+	switch (tier) {
+	case 1:
+		err = mq_ws_1_produce(message_producer_1, tmpstr, &cb_err);
+		break;
+	case 2:
+		err = mq_ws_2_produce(message_producer_2, tmpstr, &cb_err);
+		break;
+	case 3:
+		err = mq_ws_3_produce(message_producer_3, tmpstr, &cb_err);
+		break;
+	case 4:
+		err = mq_ws_4_produce(message_producer_4, tmpstr, &cb_err);
+		break;
+	case 5:
+	default:
+		err = mq_ws_5_produce(message_producer_5, tmpstr, &cb_err);
+		break;
+	}
+
+	free(tmpstr);
+	return err;
+}
+
+/* Send config/control messages to tier 5 (guaranteed delivery) */
 int jt_srv_send(int msg_type, void *msg_data)
 {
 	char *tmpstr;
@@ -155,8 +224,8 @@ int jt_srv_send(int msg_type, void *msg_data)
 		return -1;
 	}
 
-	/* write the json string to a websocket message */
-	err = mq_ws_produce(message_producer, tmpstr, &cb_err);
+	/* Config messages always go to tier 5 (lowest rate, guaranteed) */
+	err = mq_ws_5_produce(message_producer_5, tmpstr, &cb_err);
 	free(tmpstr);
 	return err;
 }
@@ -429,7 +498,8 @@ static int stats_consumer(struct mq_stats_msg *m, void *data)
 {
 	struct jt_msg_stats *s = (struct jt_msg_stats *)data;
 	mq_stats_msg_to_jt_msg_stats(m, s);
-	if (0 == jt_srv_send(JT_MSG_STATS_V1, s)) {
+	/* Route to tiered queue based on interval */
+	if (0 == jt_srv_send_tiered(JT_MSG_STATS_V1, s, m->interval_ns)) {
 		return 0;
 	}
 	return 1;
@@ -459,7 +529,8 @@ static int tt_consumer(struct mq_tt_msg *m, void *data)
 
 	//jt_messages[JT_MSG_TOPTALK_V1].print(m);
 
-	if (0 == jt_srv_send(JT_MSG_TOPTALK_V1, (struct jt_msg_toptalk *)m)) {
+	/* Route to tiered queue based on interval */
+	if (0 == jt_srv_send_tiered(JT_MSG_TOPTALK_V1, &m->m, m->m.interval_ns)) {
 		return 0;
 	}
 	return 1;
@@ -487,10 +558,7 @@ static int jt_init(void)
 		return -1;
 	}
 
-	err = mq_ws_init("ws");
-	if (err) {
-		return -1;
-	}
+	/* mq_ws is initialized in main() before WebSocket loop starts */
 
 	iface = malloc(MAX_IFACE_LEN);
 	get_first_iface(iface);
@@ -509,6 +577,12 @@ static int jt_init(void)
 	assert(!err);
 
 	g_jt_state = JT_STATE_RUNNING;
+
+	/* Drain any messages that accumulated during init.
+	 * This prevents queue overflow between thread start and first tick. */
+	jt_srv_send_stats();
+	jt_srv_send_tt();
+
 	return 0;
 }
 
@@ -532,6 +606,18 @@ int jt_srv_resume(void)
 {
 	int err;
 
+	if (JT_STATE_STOPPED == g_jt_state) {
+		/* First client connected - mq_ws already initialized in main(),
+		 * trigger full init on next tick */
+		g_jt_state = JT_STATE_STARTING;
+		return 0;
+	}
+
+	if (JT_STATE_STARTING == g_jt_state) {
+		/* Initialization in progress, will complete on next tick */
+		return 0;
+	}
+
 	if (JT_STATE_PAUSED == g_jt_state) {
 		err = mq_stats_consumer_subscribe(&stats_consumer_id);
 		assert(!err);
@@ -541,10 +627,15 @@ int jt_srv_resume(void)
 
 		g_jt_state = JT_STATE_RUNNING;
 	}
-	assert(JT_STATE_RUNNING == g_jt_state);
+
 	return 0;
 }
 
+int jt_srv_is_ready(void)
+{
+	return (g_jt_state == JT_STATE_RUNNING || g_jt_state == JT_STATE_PAUSED);
+}
+
 /* Counter for periodic pcap status updates */
 static int pcap_status_tick = 0;
 #define PCAP_STATUS_INTERVAL 100  /* Send pcap status every 100 ticks (~1s) */
@@ -575,6 +666,10 @@ int jt_server_tick(void)
 		break;
 	case JT_STATE_PAUSED:
 		break;
+	case JT_STATE_STOPPED:
+	case JT_STATE_STOPPING:
+		/* Do nothing - waiting for client or shutting down */
+		break;
 	}
 	return 0;
 }