feat: stop workers for outdated ctlogs

related to #41 and #42

Author: d-Rickyy-b

internal/certificatetransparency/ct-watcher.go

@@ -115,8 +115,6 @@ func (w *Watcher) addNewlyAvailableLogs() {
 				}
 			}
 
-			// TODO maybe add a check for logs that are still watched but no longer on the logList and remove them? See also issue #41 and #42
-
 			// If the log is not being watched, create a new worker
 			if !alreadyWatched {
 				w.wg.Add(1)
@@ -141,6 +139,33 @@ func (w *Watcher) addNewlyAvailableLogs() {
 
 	log.Printf("New ct logs found: %d\n", newCTs)
 	log.Printf("Currently monitored ct logs: %d\n", len(w.workers))
+
+	// Iterate over all workers and check if they are still in the logList
+	// If they are not, the CT Logs are probably no longer relevant.
+	// We should stop the worker if that didn't already happen.
+	for _, ctWorker := range w.workers {
+		workerURL := normalizeCtlogURL(ctWorker.ctURL)
+
+		onLogList := false
+		for _, operator := range logList.Operators {
+			// Iterate over each log of the operator
+			for _, transparencyLog := range operator.Logs {
+				// Check if the log is already being watched
+
+				logListURL := normalizeCtlogURL(transparencyLog.URL)
+				if workerURL == logListURL {
+					onLogList = true
+					break
+				}
+			}
+		}
+
+		// If the log is not in the loglist, stop the worker
+		if !onLogList {
+			log.Printf("Stopping worker. CT URL not found in LogList: '%s'\n", ctWorker.ctURL)
+			ctWorker.stop()
+		}
+	}
 }
 
 // Stop stops the watcher.
@@ -157,10 +182,13 @@ type worker struct {
 	entryChan    chan models.Entry
 	mu           sync.Mutex
 	running      bool
+	cancel       context.CancelFunc
 }
 
 // startDownloadingCerts starts downloading certificates from the CT log. This method is blocking.
 func (w *worker) startDownloadingCerts(ctx context.Context) {
+	ctx, w.cancel = context.WithCancel(ctx)
+
 	// Normalize CT URL. We remove trailing slashes and prepend "https://" if it's not already there.
 	w.ctURL = strings.TrimRight(w.ctURL, "/")
 	if !strings.HasPrefix(w.ctURL, "https://") && !strings.HasPrefix(w.ctURL, "http://") {
@@ -179,13 +207,15 @@ func (w *worker) startDownloadingCerts(ctx context.Context) {
 	}
 
 	w.running = true
+	defer func() { w.running = false }()
 	w.mu.Unlock()
 
 	for {
 		log.Printf("Starting worker for CT log: %s\n", w.ctURL)
 		workerErr := w.runWorker(ctx)
 		if workerErr != nil {
 			if errors.Is(workerErr, errFetchingSTHFailed) {
+				// TODO this could happen due to a 429 error. We should retry the request
 				log.Printf("Worker for '%s' failed - could not fetch STH\n", w.ctURL)
 				return
 			} else if errors.Is(workerErr, errCreatingClient) {
@@ -213,6 +243,13 @@ func (w *worker) startDownloadingCerts(ctx context.Context) {
 	}
 }
 
+func (w *worker) stop() {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	w.cancel()
+}
+
 // runWorker runs a single worker for a single CT log. This method is blocking.
 func (w *worker) runWorker(ctx context.Context) error {
 	hc := http.Client{Timeout: 30 * time.Second}
@@ -224,6 +261,7 @@ func (w *worker) runWorker(ctx context.Context) error {
 
 	sth, getSTHerr := jsonClient.GetSTH(ctx)
 	if getSTHerr != nil {
+		// TODO this can happen due to a 429 error. We should retry the request
 		log.Printf("Could not get STH for '%s': %s\n", w.ctURL, getSTHerr)
 		return errFetchingSTHFailed
 	}