Periodic sqlite vacuuming for lq queue

e2e/test/lqvacuumer/config.toml

@@ -0,0 +1 @@
+max-retry = 0 # avoid waiting for retries to speed up test

e2e/test/lqvacuumer/lq_vacuumer_test.go

@@ -0,0 +1,58 @@
+package nxdomain
+
+import (
+	_ "embed"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/internetarchive/Zeno/e2e"
+	"github.com/internetarchive/Zeno/internal/pkg/source/lq"
+)
+
+type recordMatcher struct {
+	vacuumedSuccess bool
+	unexpectedError bool
+}
+
+func (rm *recordMatcher) Match(record map[string]string) {
+	if record["level"] == "INFO" && record["msg"] == "vacuuming complete" {
+		rm.vacuumedSuccess = true
+	}
+	if record["level"] == "ERROR" {
+		if strings.Contains(record["err"], "failed to resolve DNS") {
+		} else {
+			rm.unexpectedError = true
+		}
+	}
+}
+
+func (rm *recordMatcher) Assert(t *testing.T) {
+	if rm.unexpectedError {
+		t.Error("An unexpected error was logged during the test")
+	}
+}
+
+func (rm *recordMatcher) ShouldStop() bool {
+	return rm.vacuumedSuccess || rm.unexpectedError
+}
+
+func Test_LQ_vacuumer(t *testing.T) {
+	os.RemoveAll("jobs")
+
+	rm := &recordMatcher{}
+	wg := &sync.WaitGroup{}
+	shouldStopCh := make(chan struct{})
+	wg.Add(2)
+
+	lq.VacuumInterval = 1 * time.Second
+
+	go e2e.StartHandleLogRecord(t, wg, rm, shouldStopCh)
+	go e2e.ExecuteCmdZenoGetURL(t, wg, []string{"http://nxdomain.nxtld/"})
+
+	e2e.WaitForGoroutines(t, wg, shouldStopCh)
+	rm.Assert(t)
+
+}

internal/pkg/source/lq/consumer.go

@@ -57,8 +57,6 @@ func (s *LQ) consumer() {
 			// Close the urlBuffer to signal consumerSenders to finish
 			close(urlBuffer)
 
-			s.wg.Done()
-
 			logger.Debug("closed")
 			return
 		}

internal/pkg/source/lq/finisher.go

@@ -59,8 +59,6 @@ func (s *LQ) finisher() {
 			// Close the batch channel to signal the dispatcher to finish.
 			close(batchCh)
 
-			s.wg.Done()
-
 			logger.Debug("closed")
 			return
 		}

internal/pkg/source/lq/lq.go

@@ -54,10 +54,10 @@ func (s *LQ) Start(finishChan, produceChan chan *models.Item) error {
 		s.produceCh = produceChan
 		s.client = LQclient
 
-		s.wg.Add(3)
-		go s.consumer()
-		go s.producer()
-		go s.finisher()
+		s.wg.Go(s.consumer)
+		s.wg.Go(s.producer)
+		s.wg.Go(s.finisher)
+		s.wg.Go(s.vacuumer)
 
 		logger.Info("started")
 

internal/pkg/source/lq/producer.go

@@ -54,8 +54,6 @@ func (s *LQ) producer() {
 			// Close the batch channel to signal the dispatcher to finish.
 			close(batchCh)
 
-			s.wg.Done()
-
 			logger.Debug("closed")
 			return
 		}

internal/pkg/source/lq/vacuumer.go

@@ -0,0 +1,39 @@
+package lq
+
+import (
+	"context"
+	"time"
+
+	"github.com/internetarchive/Zeno/internal/pkg/log"
+)
+
+var VacuumInterval = 10 * time.Minute
+
+func (s *LQ) vacuumer() {
+	logger := log.NewFieldedLogger(&log.Fields{
+		"component": "lq.vacuumer",
+	})
+
+	// Create a context to manage goroutines
+	ctx, cancel := context.WithCancel(s.ctx)
+	defer cancel()
+
+	ticker := time.NewTicker(VacuumInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Debug("context cancelled, exiting")
+			return
+		case <-ticker.C:
+			logger.Info("vacuuming")
+			_, err := s.client.dbWrite.Exec("VACUUM;")
+			if err != nil {
+				logger.Error("vacuuming failed", err)
+			}
+			logger.Info("vacuuming complete")
+		}
+	}
+
+}