Keep in sync server.System.Inflight

server/internal/clients/clients.go

@@ -573,13 +573,17 @@ func (i *Inflight) Delete(key uint16) bool {
 }
 
 // ClearExpired deletes any inflight messages that have remained longer than
-// the servers InflightTTL duration.
-func (i *Inflight) ClearExpired(expiry int64) {
+// the servers InflightTTL duration. Returns number of deleted inflights.
+func (i *Inflight) ClearExpired(expiry int64) int64 {
 	i.Lock()
 	defer i.Unlock()
+	var deleted int64
 	for k, m := range i.internal {
 		if m.Created < expiry || m.Created == 0 {
 			delete(i.internal, k)
+			deleted++
 		}
 	}
+
+	return deleted
 }

server/internal/clients/clients_test.go

@@ -916,13 +916,14 @@ func TestInflightClearExpired(t *testing.T) {
 
 	require.Len(t, cl.Inflight.internal, 4)
 
-	cl.Inflight.ClearExpired(n - 2)
+	deleted := cl.Inflight.ClearExpired(n - 2)
 	cl.Inflight.RLock()
 	defer cl.Inflight.RUnlock()
 	require.Len(t, cl.Inflight.internal, 2)
 	require.Equal(t, (n - 1), cl.Inflight.internal[1].Created)
 	require.Equal(t, (n - 2), cl.Inflight.internal[2].Created)
 	require.Equal(t, int64(0), cl.Inflight.internal[3].Created)
+	require.Equal(t, int64(2), deleted)
 }
 
 var (

server/server.go

@@ -1099,7 +1099,8 @@ func (s *Server) clearExpiredInflights(dt int64) {
 	expiry := dt - s.Options.InflightTTL
 
 	for _, client := range s.Clients.GetAll() {
-		client.Inflight.ClearExpired(expiry)
+		deleted := client.Inflight.ClearExpired(expiry)
+		atomic.AddInt64(&s.System.Inflight, deleted*-1)
 	}
 
 	if s.Store != nil {
@@ -1111,6 +1112,7 @@ func (s *Server) clearExpiredInflights(dt int64) {
 func (s *Server) clearAbandonedInflights(cl *clients.Client) {
 	for i := range cl.Inflight.GetAll() {
 		cl.Inflight.Delete(i)
+		atomic.AddInt64(&s.System.Inflight, -1)
 	}
 }
 

server/server_test.go

@@ -2739,6 +2739,7 @@ func TestServerClearExpiredInflights(t *testing.T) {
 	require.Len(t, cl.Inflight.GetAll(), 4)
 	s.clearExpiredInflights(n)
 	require.Len(t, cl.Inflight.GetAll(), 2)
+	require.Equal(t, int64(-2), s.System.Inflight)
 }
 
 func TestServerClearAbandonedInflights(t *testing.T) {
@@ -2774,4 +2775,5 @@ func TestServerClearAbandonedInflights(t *testing.T) {
 	s.clearAbandonedInflights(cl)
 	require.Len(t, cl.Inflight.GetAll(), 0)
 	require.Len(t, cl2.Inflight.GetAll(), 2)
+	require.Equal(t, int64(-2), s.System.Inflight)
 }