data-mesher/pkg/state/sweeper.go

package state

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"time"

	"git.clan.lol/clan/data-mesher/pkg/crypto"
	"git.clan.lol/clan/data-mesher/pkg/model"
)

// expiredEntry holds enough state to remove an expired sig + its on-disk file
// in the second pass of SweepExpired. We collect these during iteration so we
// don't mutate the BoltDB store while a cursor is still walking it.
type expiredEntry struct {
	networkID *crypto.PublicKey
	name      string
	filePath  string
}

// SweepExpired walks the signature store and removes any signatures whose
// ValidFor TTL has elapsed (i.e. SignedAt + ValidFor is before now), along
// with their on-disk file. Returns the number of entries removed.
//
// Expiry is decided independently by every node from signed metadata, so the
// sweeper does NOT produce a tombstone — there is nothing to gossip. A
// concurrent reader on another goroutine will either see the old signature
// and fail to open the file (returning ErrFileNotFound, which is correct
// because it is expired) or fail to find the signature at all (also
// correct).
func (f *Files) SweepExpired(now time.Time) (int, error) {
	tx, err := f.signatures.BeginTx(true)
	if err != nil {
		return 0, fmt.Errorf("failed to begin write transaction: %w", err)
	}

	defer func() {
		_ = tx.Rollback()
	}()

	r, err := f.ListSignatures(tx)
	if err != nil {
		return 0, fmt.Errorf("failed to list signatures: %w", err)
	}

	// Pass 1: collect expired entries.
	// We don't delete inside this loop because BoltDB cursors are invalidated by writes through the same tx.
	var expired []expiredEntry

	buf := make([]model.Signature, 16)

	for {
		n, readErr := r.Read(buf)

		for i := range n {
			sig := buf[i]

			if !sig.Expired(now) {
				continue
			}

			// look up the network so we can derive the on-disk file path
			network, ok := f.cfg.NetworksByID[sig.NetworkID.String()]
			if !ok {
				slog.Warn(
					"sweeper: signature for unknown network, skipping",
					slog.String("network_id", sig.NetworkID.String()),
					slog.String("name", sig.Name),
				)

				continue
			}

			filePath, fpErr := f.FilePath(network, sig.Name)
			if fpErr != nil {
				slog.Warn(
					"sweeper: failed to resolve file path, skipping",
					slog.String("name", sig.Name),
					slog.Any("err", fpErr),
				)

				continue
			}

			expired = append(expired, expiredEntry{
				networkID: sig.NetworkID,
				name:      sig.Name,
				filePath:  filePath,
			})
		}

		if errors.Is(readErr, io.EOF) {
			break
		}

		if readErr != nil {
			return 0, fmt.Errorf("failed to read signatures: %w", readErr)
		}
	}

	if len(expired) == 0 {
		// nothing to do — release the (still-empty) write tx
		return 0, nil
	}

	// Pass 2: delete signatures from the DB, then commit, then remove files from disk.
	// Doing things in this order ensures the only possible inconsistency is orphaned files on disk with no signatures
	// in the db, effectively making them invisible.
	// The integrity checker will clean up any orphans on next startup.
	for _, e := range expired {
		if err = f.signatures.Delete(tx, e.networkID, e.name); err != nil {
			return 0, fmt.Errorf("failed to delete signature %s: %w", e.name, err)
		}
	}

	if err = tx.Commit(); err != nil {
		return 0, fmt.Errorf("failed to commit sweep transaction: %w", err)
	}

	// Pass 3: best-effort file removal.
	// Failures are logged but don't prevent us from reporting how many entries were swept — the DB is already
	// consistent and orphaned files are harmless.
	for _, e := range expired {
		if err = os.Remove(e.filePath); err != nil && !errors.Is(err, os.ErrNotExist) {
			slog.Warn(
				"sweeper: failed to remove expired file",
				slog.String("path", e.filePath),
				slog.Any("err", err),
			)
		}
	}

	return len(expired), nil
}

// SweepLoop runs SweepExpired on the given interval until ctx is cancelled.
// Errors from individual sweeps are logged but not returned, so a transient
// failure does not bring down the server's errgroup. Returns nil on
// context cancellation.
func (f *Files) SweepLoop(ctx context.Context, interval time.Duration, clock model.Clock) error {
	if interval <= 0 {
		slog.Info("sweeper disabled (interval <= 0)")
		<-ctx.Done()

		return nil
	}

	if clock == nil {
		return errors.New("clock cannot be nil")
	}

	slog.Info("starting expired-file sweeper", slog.Duration("interval", interval))

	t := time.NewTicker(interval)
	defer t.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil

		case <-t.C:
			n, err := f.SweepExpired(clock())
			if err != nil {
				slog.Error("sweep failed", slog.Any("err", err))

				continue
			}

			if n > 0 {
				slog.Info("swept expired files", slog.Int("removed", n))
			}
		}
	}
}