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 }