feat: add diagnostic logging and stats for direct mode connections

Add per-connection close logging (DC number, duration, which side closed)
and two new aggregate counters for monitoring direct mode health:
- direct_dc_connections_failed: connections that failed to establish
- direct_dc_connections_dc_closed: connections closed by the DC side

Exposed in plain text stats, Prometheus metrics, and verbose logs.
Helps diagnose incomplete media loading reported in #64.

Author: dvershinin

mtproto/mtproto-proxy.c

@@ -407,6 +407,7 @@ struct worker_stats {
   long long tot_forwarded_simple_acks, dropped_simple_acks;
   long long mtproto_proxy_errors;
   long long direct_dc_connections_created, direct_dc_connections_active;
+  long long direct_dc_connections_failed, direct_dc_connections_dc_closed;
 
   long long connections_failed_lru, connections_failed_flood;
 
@@ -433,6 +434,7 @@ long long tot_forwarded_responses, dropped_responses;
 long long tot_forwarded_simple_acks, dropped_simple_acks;
 long long mtproto_proxy_errors;
 long long direct_dc_connections_created, direct_dc_connections_active;
+long long direct_dc_connections_failed, direct_dc_connections_dc_closed;
 
 char proxy_tag[16];
 int proxy_tag_set;
@@ -466,10 +468,12 @@ static void update_local_stats_copy (struct worker_stats *S) {
   UPD (mtproto_proxy_errors);
   UPD (direct_dc_connections_created);
   UPD (direct_dc_connections_active);
+  UPD (direct_dc_connections_failed);
+  UPD (direct_dc_connections_dc_closed);
   UPD (connections_failed_lru);
   UPD (connections_failed_flood);
-  UPD (ext_connections); 
-  UPD (ext_connections_created); 
+  UPD (ext_connections);
+  UPD (ext_connections_created);
   UPD (http_queries);
   UPD (http_bad_headers);
   { int _i; for (_i = 0; _i < 16; _i++) {
@@ -547,6 +551,8 @@ static inline void add_stats (struct worker_stats *W) {
   UPD (mtproto_proxy_errors);
   UPD (direct_dc_connections_created);
   UPD (direct_dc_connections_active);
+  UPD (direct_dc_connections_failed);
+  UPD (direct_dc_connections_dc_closed);
   UPD (connections_failed_lru);
   UPD (connections_failed_flood);
   UPD (ext_connections);
@@ -680,6 +686,8 @@ void mtfront_prepare_stats (stats_buffer_t *sb) {
 	     "direct_mode\t%d\n"
 	     "direct_dc_connections_created\t%lld\n"
 	     "direct_dc_connections_active\t%lld\n"
+	     "direct_dc_connections_failed\t%lld\n"
+	     "direct_dc_connections_dc_closed\t%lld\n"
 	     "version\t" VERSION_STR " compiled at " __DATE__ " " __TIME__ " by gcc " __VERSION__ " "
 #ifdef __LP64__
 	     "64-bit"
@@ -749,7 +757,9 @@ void mtfront_prepare_stats (stats_buffer_t *sb) {
 	     proxy_tag_set,
 	     direct_mode,
 	     S(direct_dc_connections_created),
-	     S(direct_dc_connections_active)
+	     S(direct_dc_connections_active),
+	     S(direct_dc_connections_failed),
+	     S(direct_dc_connections_dc_closed)
   );
 
   { int _sc = tcp_rpcs_get_ext_secret_count();
@@ -843,7 +853,13 @@ void mtfront_prepare_prometheus_stats (stats_buffer_t *sb) {
 	     "mtproxy_ip_acl_rejected_total %lld\n"
 	     "# HELP mtproxy_direct_dc_connections_created_total Direct DC connections created.\n"
 	     "# TYPE mtproxy_direct_dc_connections_created_total counter\n"
-	     "mtproxy_direct_dc_connections_created_total %lld\n",
+	     "mtproxy_direct_dc_connections_created_total %lld\n"
+	     "# HELP mtproxy_direct_dc_connections_failed_total Direct DC connections that failed to establish.\n"
+	     "# TYPE mtproxy_direct_dc_connections_failed_total counter\n"
+	     "mtproxy_direct_dc_connections_failed_total %lld\n"
+	     "# HELP mtproxy_direct_dc_connections_dc_closed_total Direct DC connections closed by the DC side.\n"
+	     "# TYPE mtproxy_direct_dc_connections_dc_closed_total counter\n"
+	     "mtproxy_direct_dc_connections_dc_closed_total %lld\n",
 	     S(get_queries),
 	     S(tot_forwarded_queries),
 	     S(expired_forwarded_queries),
@@ -862,7 +878,9 @@ void mtfront_prepare_prometheus_stats (stats_buffer_t *sb) {
 	     S(http_queries),
 	     S(http_bad_headers),
 	     S(conn.accept_ip_acl_rejected),
-	     S(direct_dc_connections_created)
+	     S(direct_dc_connections_created),
+	     S(direct_dc_connections_failed),
+	     S(direct_dc_connections_dc_closed)
   );
 
   /* gauges */

net/net-tcp-rpc-ext-server.c

@@ -178,6 +178,7 @@ int tcp_proxy_pass_write_packet (connection_job_t C, struct raw_message *raw) {
 extern int direct_mode;
 extern int workers;
 extern long long direct_dc_connections_created, direct_dc_connections_active;
+extern long long direct_dc_connections_failed, direct_dc_connections_dc_closed;
 extern long long per_secret_connections[16], per_secret_connections_created[16];
 extern long long per_secret_connections_rejected[16];
 
@@ -278,14 +279,26 @@ static int tcp_direct_dc_parse_execute (connection_job_t C) {
 
 static int tcp_direct_close (connection_job_t C, int who) {
   struct connection_info *c = CONN_INFO(C);
-  vkprintf (1, "closing direct connection #%d %s:%d -> %s:%d\n", c->fd, show_our_ip (C), c->our_port, show_remote_ip (C), c->remote_port);
-  if ((c->type == &ct_direct_client || c->type == &ct_direct_client_drs) && direct_dc_connections_active > 0) {
+  int is_client = (c->type == &ct_direct_client || c->type == &ct_direct_client_drs);
+  int is_dc = (c->type == &ct_direct_dc);
+  int target_dc = TCP_RPC_DATA(C)->extra_int4;
+  double duration = precise_now - c->query_start_time;
+
+  vkprintf (1, "direct: closing %s connection #%d (DC %d) after %.1fs, %s:%d -> %s:%d, who=%d\n",
+            is_client ? "client" : "DC", c->fd, target_dc, duration,
+            show_our_ip (C), c->our_port, show_remote_ip (C), c->remote_port, who);
+
+  if (is_client && direct_dc_connections_active > 0) {
     direct_dc_connections_active--;
     int sid = TCP_RPC_DATA(C)->extra_int2;
     if (sid > 0 && sid <= 16) {
       per_secret_connections[sid - 1]--;
     }
   }
+  if (is_dc && who != 0) {
+    /* DC side closed unexpectedly (not by us tearing down the pair) */
+    direct_dc_connections_dc_closed++;
+  }
   if (c->extra) {
     job_t E = PTR_MOVE (c->extra);
     fail_connection (E, -23);
@@ -389,7 +402,8 @@ static int direct_connect_to_dc (connection_job_t C, int target_dc) {
 
   const struct dc_address *dc = direct_dc_lookup (target_dc);
   if (!dc) {
-    vkprintf (1, "direct mode: unknown DC %d, closing connection\n", target_dc);
+    kprintf ("direct mode: unknown DC %d, closing connection\n", target_dc);
+    direct_dc_connections_failed++;
     fail_connection (C, -1);
     return 0;
   }
@@ -408,6 +422,7 @@ static int direct_connect_to_dc (connection_job_t C, int target_dc) {
   int cfd = client_socket (dc->ipv4, dc->port, 0);
   if (cfd < 0) {
     kprintf ("direct mode: failed to connect to DC %d: %m\n", target_dc);
+    direct_dc_connections_failed++;
     fail_connection (C, -27);
     return 0;
   }
@@ -417,7 +432,8 @@ static int direct_connect_to_dc (connection_job_t C, int target_dc) {
                                     ntohl (dc->ipv4), NULL, dc->port);
 
   if (!EJ) {
-    kprintf ("direct mode: failed to create DC connection\n");
+    kprintf ("direct mode: failed to create DC connection for DC %d\n", target_dc);
+    direct_dc_connections_failed++;
     job_decref_f (C);
     fail_connection (C, -37);
     return 0;