Keep the values of numbered traffic counters across reloads

What a numbered counter counted now belongs to the process: a counter
rule points to the value of its number, created the first time the
number is met (from the counter file, if one is open) and shared by the
rules of that number in every configuration loaded. A reload no longer
resets it when there is no counter file, and once configurations can
outlive one another, clients of the old and the new one count into the
same value. An unnumbered counter, and those LdapPlugin makes, keep a
value of their own as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/3proxy.c

@@ -214,7 +214,7 @@ void dumpcounters(struct trafcount *tlin, int counterd){
 		if(cfp){
 			for(tl = tlin; cfp && tl; tl = tl->next){
 				if(tl->type >= conf.countertype)
-					fprintf(cfp, "%05d %020"PRIu64"%s%s\n", tl->number, tl->traf64, tl->comment?" #" : "", tl->comment? tl->comment : "");
+					fprintf(cfp, "%05d %020"PRIu64"%s%s\n", tl->number, tl->v->traf64, tl->comment?" #" : "", tl->comment? tl->comment : "");
 			}
 			fclose(cfp);
 		}
@@ -228,15 +228,15 @@ void dumpcounters(struct trafcount *tlin, int counterd){
 			if(lseek(counterd, 
 				sizeof(struct counter_header) + (tl->number - 1) * sizeof(struct counter_record),
 				SEEK_SET) >= 0){
-			    crecord.traf64 = tl->traf64;
-			    crecord.cleared = tl->cleared;
-			    crecord.updated = tl->updated;
+			    crecord.traf64 = tl->v->traf64;
+			    crecord.cleared = tl->v->cleared;
+			    crecord.updated = tl->v->updated;
 			    if(write(counterd, &crecord, sizeof(struct counter_record))){}
 			}
 		}
-		if(tl->type!=NEVER && timechanged(tl->cleared, curtime, tl->type)){
-			tl->cleared = curtime;
-			tl->traf64 = 0;
+		if(tl->type!=NEVER && timechanged(tl->v->cleared, curtime, tl->type)){
+			tl->v->cleared = curtime;
+			tl->v->traf64 = 0;
 		}
 	}
  }

src/auth.c

@@ -47,12 +47,12 @@ int alwaysauth(struct clientparam * param){
 						countout = 1;
 						if(tc->ace->action != COUNTALL) continue;
 					}
-					if(tc->traflim64 <= tc->traf64) {
+					if(tc->traflim64 <= tc->v->traf64) {
 					    _3proxy_mutex_unlock(&tc_mutex);
 					    return 10;
 					}
 					param->trafcountfunc = conf.trafcountfunc;
-					param->maxtrafin64 = tc->traflim64 - tc->traf64;
+					param->maxtrafin64 = tc->traflim64 - tc->v->traf64;
 				}
 			}
 			if(countout)for(tc = conf.trafcounter; tc; tc = tc->next) {
@@ -62,12 +62,12 @@ int alwaysauth(struct clientparam * param){
 					if(tc->ace->action != COUNTOUT && tc->ace->action !=  COUNTALL) {
 						continue;
 					}
-					if(tc->traflim64 <= tc->traf64) {
+					if(tc->traflim64 <= tc->v->traf64) {
 					    _3proxy_mutex_unlock(&tc_mutex);
 					    return 10;
 					}
 					param->trafcountfunc = conf.trafcountfunc;
-					param->maxtrafout64 = tc->traflim64 - tc->traf64;
+					param->maxtrafout64 = tc->traflim64 - tc->v->traf64;
 				}
 			}
 			_3proxy_mutex_unlock(&tc_mutex);

src/conf.c

@@ -47,6 +47,42 @@ int linenum = 0;
 struct counter_header cheader = {"3CF", (time_t)0};
 struct counter_record crecord;
 
+/* The values of the numbered counters, process-wide: a counter of a number
+   met for the first time takes its value from the counter file. Only the
+   thread reading the configuration adds to the list. */
+static struct trafcell {
+	struct trafcell *next;
+	unsigned number;
+	struct trafvalue v;
+} *trafcells = NULL;
+
+struct trafvalue * trafvalue(unsigned number){
+	struct trafcell *tc;
+
+	for(tc = trafcells; tc; tc = tc->next) if(tc->number == number) return &tc->v;
+	if(!(tc = malloc(sizeof(*tc)))) return NULL;
+	memset(tc, 0, sizeof(*tc));
+	tc->number = number;
+	if(conf.counterd >= 0) {
+		lseek(conf.counterd,
+			sizeof(struct counter_header) + (number - 1) * sizeof(struct counter_record),
+			SEEK_SET);
+		memset(&crecord, 0, sizeof(struct counter_record));
+		if(read(conf.counterd, &crecord, sizeof(struct counter_record)) == sizeof(struct counter_record)){
+			if(crecord.cleared < 0 || crecord.cleared >= MAX_COUNTER_TIME || crecord.updated < 0 || crecord.updated >= MAX_COUNTER_TIME){
+				free(tc);
+				return NULL;
+			}
+			tc->v.traf64 = crecord.traf64;
+			tc->v.cleared = crecord.cleared;
+			tc->v.updated = crecord.updated;
+		}
+	}
+	tc->next = trafcells;
+	trafcells = tc;
+	return &tc->v;
+}
+
 int mainprepare(int argc, char** argv, void **service);
 int mainrun(void *service);
 void maindrop(void *service);
@@ -1718,22 +1754,16 @@ static int h_ace(int argc, unsigned char **argv){
 				fprintf(stderr, "Wrong traffic limit specified, line %d\n", linenum);
 				return(6);
 			}
-			if(tl->number != 0 && conf.counterd >= 0) {
-				lseek(conf.counterd, 
-					sizeof(struct counter_header) + (tl->number - 1) * sizeof(struct counter_record),
-					SEEK_SET);
-				memset(&crecord, 0, sizeof(struct counter_record));
-				if(read(conf.counterd, &crecord, sizeof(struct counter_record)) == sizeof(struct counter_record)){
-				    tl->traf64 = crecord.traf64;
-				    tl->cleared = crecord.cleared;
-				    tl->updated = crecord.updated;
-				    if(tl->cleared < 0 || tl->cleared >=  MAX_COUNTER_TIME || tl->updated < 0 || tl->updated >=  MAX_COUNTER_TIME){
-					    fprintf(stderr, "Invalid, incompatible or corrupted counter file.\n");
-					    return(6);
-				    }
-				}
+			if(!tl->number) tl->v = &tl->own;
+			else if(!(tl->v = trafvalue(tl->number))){
+				free(tl->comment);
+				free(tl);
+				freeacl(acl);
+				fprintf(stderr, "Invalid, incompatible or corrupted counter file.\n");
+				return(6);
 			}
 		}
+		else tl->v = &tl->own;
 		_3proxy_mutex_lock(&tc_mutex);
 		if(!conf.trafcounter){
 			conf.trafcounter = tl;

src/datatypes.c

@@ -471,7 +471,7 @@ static void * ef_trafcounter_type(struct node * node){
 }
 
 static void * ef_trafcounter_traffic64(struct node * node){
-	return &((struct trafcount *)node->value) -> traf64;
+	return &((struct trafcount *)node->value) -> v->traf64;
 }
 static void * ef_trafcounter_limit64(struct node * node){
 	return &((struct trafcount *)node->value) -> traflim64;
@@ -493,11 +493,11 @@ static void * ef_client_bytesout64(struct node * node){
 }
 
 static void * ef_trafcounter_cleared(struct node * node){
-	return &((struct trafcount *)node->value) -> cleared;
+	return &((struct trafcount *)node->value) -> v->cleared;
 }
 
 static void * ef_trafcounter_updated(struct node * node){
-	return &((struct trafcount *)node->value) -> updated;
+	return &((struct trafcount *)node->value) -> v->updated;
 }
 
 static void * ef_trafcounter_comment(struct node * node){

src/limiter.c

@@ -184,8 +184,8 @@ void trafcountfunc(struct clientparam *param){
 				countout = 1;
 				continue;
 			}
-			tc->traf64 += param->statssrv64;
-			tc->updated = curtime;
+			tc->v->traf64 += param->statssrv64;
+			tc->v->updated = curtime;
 		}
 	}
 	if(countout) for(tc = conf.trafcounter; tc; tc = tc->next) {
@@ -194,8 +194,8 @@ void trafcountfunc(struct clientparam *param){
 			if(tc->ace->action != COUNTOUT && tc->ace->action != COUNTALL ) {
 				continue;
 			}
-			tc->traf64 += param->statscli64;
-			tc->updated = curtime;
+			tc->v->traf64 += param->statscli64;
+			tc->v->updated = curtime;
 		}
 	}
 

src/plugins/LdapPlugin/ldapauth.c

@@ -120,8 +120,8 @@ int savecounters(void *v)
       sprintf(tmpbuf,pat_file,ldap_dircount,tcd->ace->users->user);
       f=fopen(tmpbuf,"w+b");
       fseek(f,0,SEEK_SET);
-      fprintf(f,"%"PRIu64" %lu %lu\n",tcd->traf64,
-					(unsigned long)tcd->cleared,(unsigned long)tcd->updated);
+      fprintf(f,"%"PRIu64" %lu %lu\n",tcd->v->traf64,
+					(unsigned long)tcd->v->cleared,(unsigned long)tcd->v->updated);
 
       fclose(f);
       free(tmpbuf);
@@ -433,6 +433,7 @@ int h_trafgroup(int argc, unsigned char ** argv)
              newtrafcount =(*mypluginlink->mallocfunc)(sizeof (struct trafcount));
              memset(newtrafcount, 0, sizeof(struct trafcount));
              newtrafcount->ace = newace;
+             newtrafcount->v = &newtrafcount->own;
              newtrafcount->type=rtype;
              newtrafcount->traflim64  = traflimit;
              newtrafcount->comment=(*mypluginlink->strdupfunc)("ldapcounters");
@@ -450,9 +451,9 @@ int h_trafgroup(int argc, unsigned char ** argv)
 				&rcounter.cleared, &rcounter.updated);
 
 
-               newtrafcount->traf64=rcounter.traf64;
-               newtrafcount->cleared=rcounter.cleared;
-               newtrafcount->updated=rcounter.updated;
+               newtrafcount->own.traf64=rcounter.traf64;
+               newtrafcount->own.cleared=rcounter.cleared;
+               newtrafcount->own.updated=rcounter.updated;
                fclose(f);
               }
              free(tmpbuf);   

src/proxy.h

@@ -291,6 +291,7 @@ int ipauth(struct clientparam * param);
 int doauth(struct clientparam * param);
 int strongauth(struct clientparam * param);
 void trafcountfunc(struct clientparam *param);
+struct trafvalue * trafvalue(unsigned number);
 unsigned bandlimitfunc(struct clientparam *param, unsigned nbytesin, unsigned nbytesout);
 int handleredirect(struct clientparam * param, struct ace * acentry);
 

src/structures.h

@@ -471,17 +471,26 @@ struct schedule {
 };
 
 
+/* What a traffic counter has counted. A numbered counter's lives as long as
+   the process, shared by the counters of that number in every configuration
+   loaded, so that a reload neither loses it nor splits it; an unnumbered
+   counter keeps its own. */
+struct trafvalue {
+	uint64_t traf64;
+	time_t cleared;
+	time_t updated;
+};
+
 struct trafcount {
 	struct trafcount *next;
 	struct ace *ace;
 	unsigned number;
 	ROTATION type;
-	uint64_t traf64;
 	uint64_t traflim64;
 	char * comment;
 	int disabled;
-	time_t cleared;
-	time_t updated;
+	struct trafvalue *v;	/* the shared value, or &own */
+	struct trafvalue own;
 };
 
 struct nserver {

src/webadmin.c

@@ -458,16 +458,16 @@ int op_admin_counters(struct httpreq *r, const unsigned char *params)
 					"<td>%s</td>",
 				 cp->traflim64 / (1024 * 1024),
 				 rotations[cp->type],
-				 cp->traf64 / (1024 * 1024),
-				 (((cp->traf64/16) *10) / (64*1024))%10,
-				 cp->cleared?ctime(&cp->cleared):"never"
+				 cp->v->traf64 / (1024 * 1024),
+				 (((cp->v->traf64/16) *10) / (64*1024))%10,
+				 cp->v->cleared?ctime(&cp->v->cleared):"never"
 				);
 			 inbuf += sprintf(buf + inbuf,
 					"<td>%s</td>"
 					"<td>%i</td>"
 					"</tr>\r\n",
 
-				 cp->updated?ctime(&cp->updated):"never",
+				 cp->v->updated?ctime(&cp->v->updated):"never",
 				 cp->number
 				);
 			 }

tests/cases/counters.py

@@ -0,0 +1,58 @@
+"""Traffic counters: a numbered counter keeps what it counted across a
+reload, the value belonging to the process rather than to the configuration
+read."""
+
+import re
+import time
+
+
+def run(t):
+    origin = t.free_port()
+    proxy = t.free_port()
+    adm = t.free_port()
+    server = t.start("counters", f"""
+        log
+        auth iponly
+        allow *
+        http data * /data
+        httpsrv -p{origin}
+
+        flush
+        auth iponly
+        allow *
+        countin 3/counted D 100 * * *
+        proxy -p{proxy}
+        admin -p{adm}
+    """, ports=[origin, proxy, adm])
+
+    def counted():
+        """The counter's value in MB, as the counters page shows it."""
+        page = t.http(f"http://127.0.0.1:{adm}/C").text
+        m = re.search(r"counted.*?</td><td>([0-9]+\.[0-9])</td>", page, re.S)
+        return float(m.group(1)) if m else None
+
+    def wait_counted(least, timeout=5.0):
+        deadline = time.time() + timeout
+        value = counted()
+        while (value is None or value < least) and time.time() < deadline:
+            time.sleep(0.1)
+            value = counted()
+        return value
+
+    reply = t.http(f"http://127.0.0.1:{origin}/data?size=300000",
+                   proxy=f"127.0.0.1:{proxy}")
+    t.eq(300000, reply.length, "a body passes through the counted proxy")
+    before = wait_counted(0.2)
+    t.ok("the counter counts it") if before is not None and before >= 0.2 else \
+        t.fail("the counter counts it", ">= 0.2", before)
+
+    # reload, and wait for the services of the new configuration to run
+    out = server.output()
+    accepting = out.count("Accepting connections")
+    t.http(f"http://127.0.0.1:{adm}/R")
+    deadline = time.time() + 10
+    while time.time() < deadline and server.output().count("Accepting connections") < accepting + 3:
+        time.sleep(0.05)
+
+    after = wait_counted(0.2)
+    t.eq(before, after, "a reload keeps what the numbered counter counted")