pixa/internal/imgcache/pendingaccesstimes.go

package imgcache

import (
	"context"
	"fmt"
	"time"

	"sneak.berlin/go/pixa/internal/imagerequest"
)

// pendingWriteInterval is how often the goroutine
// StartPendingAccessTimeWrites starts writes the pending access times.
// SQLite's CURRENT_TIMESTAMP, which an access time is written as, counts
// whole seconds.
const pendingWriteInterval = time.Second

// StartPendingAccessTimeWrites starts the goroutine that writes, every
// pendingWriteInterval, the pending access times in one transaction.
// Requests only hand these to it, so none waits for the database to record
// them, and only this goroutine's write ever waits for it. It is a no-op
// when already started. The goroutine outlives the caller, so it runs with
// its own context, which StopPendingAccessTimeWrites cancels to tell it to
// finish.
func (c *Cache) StartPendingAccessTimeWrites() {
	if c.pendingAccessTimesCancel != nil {
		return
	}

	ctx, cancel := context.WithCancel(context.Background())
	c.pendingAccessTimesCancel = cancel

	go func() {
		c.pendingAccessTimeWriteLoop(ctx)
	}()
}

// StopPendingAccessTimeWrites tells the goroutine
// StartPendingAccessTimeWrites started, if it did, to finish, and waits for
// its write under way to end and for it to return. It then writes the
// pending access times left, once, so that they are written before the
// database closes. It waits for all of this at most until ctx ends. When
// access times are not written, it logs that and returns an error.
func (c *Cache) StopPendingAccessTimeWrites(ctx context.Context) error {
	var err error

	if c.pendingAccessTimesCancel != nil {
		c.pendingAccessTimesCancel()

		select {
		case <-c.pendingAccessTimesDone:
		case <-ctx.Done():
			// The write given up on here may not reach the database:
			// database/sql closes without waiting for a query, and a write
			// still waiting for the connection fails when the database
			// closes. Nothing is written here.
			err = fmt.Errorf("access times still being written: %w", ctx.Err())
		}
	}

	if err == nil {
		err = c.writePendingAccessTimes(ctx)
	}

	if err != nil {
		c.log.Error("access times not written at shutdown", "error", err)
	}

	return err
}

// pendingAccessTimeWriteLoop is the body of the goroutine
// StartPendingAccessTimeWrites starts. Once ctx is cancelled it starts no
// write and returns, but does not cut off a write under way then. Access
// times whose write fails are dropped.
func (c *Cache) pendingAccessTimeWriteLoop(ctx context.Context) {
	defer close(c.pendingAccessTimesDone)

	ticker := time.NewTicker(pendingWriteInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}

		// select picks at random when ctx is cancelled and a tick waits
		// too.
		if ctx.Err() != nil {
			return
		}

		err := c.writePendingAccessTimes(context.WithoutCancel(ctx))
		if err != nil {
			c.log.Warn("failed to write access times", "error", err)
		}
	}
}

// writePendingAccessTimes takes out the variants waiting in
// pendingVariantAccessTimes and the sources waiting in
// pendingSourceAccessTimes, and sets their last_accessed_at to now in one
// transaction, once for each however often it was used. Nothing else takes
// from those channels while it runs: only the goroutine
// StartPendingAccessTimeWrites starts calls it, or
// StopPendingAccessTimeWrites once that goroutine has returned.
func (c *Cache) writePendingAccessTimes(ctx context.Context) error {
	variants := make(map[imagerequest.VariantKey]struct{})

	for range len(c.pendingVariantAccessTimes) {
		variants[<-c.pendingVariantAccessTimes] = struct{}{}
	}

	sources := make(map[imagerequest.SourceKey]struct{})

	for range len(c.pendingSourceAccessTimes) {
		sources[<-c.pendingSourceAccessTimes] = struct{}{}
	}

	accessTimes := len(variants) + len(sources)
	if accessTimes == 0 {
		return nil
	}

	tx, err := c.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
	}

	// Does nothing once the transaction is committed
	defer func() { _ = tx.Rollback() }()

	for cacheKey := range variants {
		_, err = tx.ExecContext(ctx, `
			UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
			WHERE cache_key = ?
		`, string(cacheKey))
		if err != nil {
			return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
		}
	}

	for sourceKey := range sources {
		_, err = tx.ExecContext(ctx, `
			UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
			WHERE source_key = ?
		`, string(sourceKey))
		if err != nil {
			return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
		}
	}

	err = tx.Commit()
	if err != nil {
		return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
	}

	return nil
}