package imgcache import ( "bytes" "context" "database/sql" "errors" "log/slog" "strings" "testing" "time" "sneak.berlin/go/pixa/internal/imagerequest" ) // pendingAccessTimesWait is how long a test waits for the database's // connection, for an access time to reach the database, for a write to start // waiting for the connection, for a request to return past its deadline, or // for StopPendingAccessTimeWrites to return. const pendingAccessTimesWait = 5 * time.Second // holdDatabase takes the one connection of cache's database, so that every // other query waits for it, and returns the func that frees it. func holdDatabase(t *testing.T, cache *Cache) func() { t.Helper() ctx, cancel := context.WithTimeout(t.Context(), pendingAccessTimesWait) defer cancel() conn, err := cache.db.Conn(ctx) if err != nil { t.Fatalf("failed to take the database connection: %v", err) } return func() { _ = conn.Close() } } // waitForConnectionWaits waits until db.Stats().WaitCount, the number of // times a caller has waited for a connection, reaches waits. func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) { t.Helper() deadline := time.Now().Add(pendingAccessTimesWait) for db.Stats().WaitCount < waits { if time.Now().After(deadline) { t.Fatalf("callers waited for the database connection %d times, want %d", db.Stats().WaitCount, waits) } time.Sleep(10 * time.Millisecond) } } // TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the // database's one connection while a request whose fetch is held reaches its // deadline. The request must return with the deadline's error while the // connection is still held, as it waits for no database write past its // deadline. func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) { t.Parallel() svc, fixtures, fetcher := setupHeldFetchService(t) svc.cache.StartPendingAccessTimeWrites() defer func() { _ = svc.cache.StopPendingAccessTimeWrites(t.Context()) }() // Room for the request to reach its fetch on a busy host const timeout = 2 * time.Second ctx, cancel := context.WithTimeout(t.Context(), timeout) defer cancel() deadline, _ := ctx.Deadline() results := startGet(ctx, svc, photoVariant(fixtures, 85, imagerequest.FitCover)) // The request has made its database reads by the time it fetches. select { case <-fetcher.started: case <-time.After(timeout): t.Fatal("request did not reach its fetch by its deadline") } releaseDatabase := holdDatabase(t, svc.cache) defer releaseDatabase() select { case got := <-results: t.Logf("Get() returned %v after its deadline, error = %v", time.Since(deadline), got.err) if !errors.Is(got.err, context.DeadlineExceeded) { t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded) } case <-time.After(time.Until(deadline) + pendingAccessTimesWait): t.Fatal("request did not return while the database was busy") } } // TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds holds the // database's one connection while the goroutine that writes the pending // access times waits for it, then stops that goroutine with a context that // ends first. The stop must return the context's error, and the write under // way must still reach the database once the connection is free. func TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds(t *testing.T) { t.Parallel() cache, _ := newEvictionTestCache(t, 1<<30) storeEvictionTestVariant(t, cache, testVariantKeyOne, []byte("variant bytes")) anHourAgo := time.Now().Add(-time.Hour) setVariantLastAccessed(t, cache, testVariantKeyOne, anHourAgo) cache.StartPendingAccessTimeWrites() defer func() { _ = cache.StopPendingAccessTimeWrites(t.Context()) }() releaseDatabase := holdDatabase(t, cache) defer releaseDatabase() waitsBefore := cache.db.Stats().WaitCount cache.touchVariant(testVariantKeyOne) // The goroutine's write waits for the connection. waitForConnectionWaits(t, cache.db, waitsBefore+1) stopCtx, cancelStop := context.WithCancel(t.Context()) cancelStop() stopped := make(chan error, 1) go func() { stopped <- cache.StopPendingAccessTimeWrites(stopCtx) }() select { case err := <-stopped: if !errors.Is(err, context.Canceled) { t.Errorf("StopPendingAccessTimeWrites() error = %v, want %v", err, context.Canceled) } case <-time.After(pendingAccessTimesWait): t.Fatal("StopPendingAccessTimeWrites() did not return once its context ended") } releaseDatabase() deadline := time.Now().Add(pendingAccessTimesWait) for countRows(t, cache, ` SELECT COUNT(*) FROM variant_content WHERE cache_key = ? AND last_accessed_at > ? `, string(testVariantKeyOne), anHourAgo.UTC().Format(sqliteTimestampFormat)) == 0 { if time.Now().After(deadline) { t.Fatal("the access time being written was not written") } time.Sleep(10 * time.Millisecond) } } // TestLookupSource_HandsAccessTimeToTheBackgroundWriter finds a cached source // whose access time is an hour old. LookupSource must not write the access // time itself but hand it over to be written, and StopPendingAccessTimeWrites // must then write it. func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) { t.Parallel() cache, _ := newEvictionTestCache(t, 1<<30) key := storeEvictionTestSource(t, cache, "example.com", "/photo.jpg", []byte("source bytes")) anHourAgo := time.Now().Add(-time.Hour) setSourceLastAccessed(t, cache, key, anHourAgo) touchedQuery := ` SELECT COUNT(*) FROM source_content WHERE source_key = ? AND last_accessed_at > ? ` anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat) found, err := cache.LookupSource(t.Context(), &imagerequest.Request{ Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"}, }) if err != nil || found != key { t.Fatalf("LookupSource() = %q, %v; want %q", found, err, key) } if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 { t.Error("LookupSource wrote the source's access time itself") } if waiting := len(cache.pendingSourceAccessTimes); waiting != 1 { t.Errorf("%d access times of sources wait to be written, want 1", waiting) } err = cache.StopPendingAccessTimeWrites(t.Context()) if err != nil { t.Fatalf("StopPendingAccessTimeWrites() error = %v", err) } if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 1 { t.Error("the source's access time was not written") } } // TestStopPendingAccessTimeWrites_LogsWhatWasNotWritten makes the write of // the access times fail at shutdown, by dropping the table they are written // to. The stop must return an error and log that access times were not // written. func TestStopPendingAccessTimeWrites_LogsWhatWasNotWritten(t *testing.T) { t.Parallel() cache, _ := newEvictionTestCache(t, 1<<30) var logs bytes.Buffer cache.log = slog.New(slog.NewJSONHandler(&logs, nil)) cache.touchVariant(testVariantKeyOne) _, err := cache.db.ExecContext(t.Context(), `DROP TABLE variant_content`) if err != nil { t.Fatalf("dropping variant_content: %v", err) } err = cache.StopPendingAccessTimeWrites(t.Context()) t.Logf("StopPendingAccessTimeWrites() error = %v; logged:\n%s", err, logs.String()) if err == nil { t.Error("StopPendingAccessTimeWrites() error = nil, want the access times' error") } if !strings.Contains(logs.String(), "access times not written at shutdown") { t.Error("the access times not written were not logged") } }