Fixed contrast, message and 404 handling

internal/datasource/biblioreads/adapter.go

@@ -14,7 +14,7 @@ const (
 	// maxConcurrentLookups bounds lookups across every enrichment caller, so they do not pile up
 	// waiting on the client. Client request spacing and cooldowns protect each instance.
 	maxConcurrentLookups = 4
-	requestTimeout       = 2 * time.Minute
+	requestTimeout       = 3 * time.Minute
 )
 
 type SearchSource interface {

internal/datasource/biblioreads/adapter_test.go

@@ -875,6 +875,36 @@ func TestTitleFallback(t *testing.T) {
 	}
 }
 
+func TestUnavailableSearchDoesNotBroadenQuery(t *testing.T) {
+	const title = "El economista camuflado"
+	const author = "Tim Harford"
+	source := &fakeSource{
+		byQuery: map[string][]SearchResult{
+			title: {{Title: title, Author: author, BookURL: "/book/show/123", Rating: "4.00"}},
+		},
+		queryErrors: map[string]error{title + " " + author: ErrUnavailable},
+	}
+	idx := newTestIndexer(t, source)
+	doc := index.Document{ID: testID, Metadata: metadata.Metadata{Title: title, Authors: []string{author}}}
+	if err := index.IndexDocument(idx.documentsIdx, doc.ID, doc); err != nil {
+		t.Fatal(err)
+	}
+	got, err := idx.Book(context.Background(), doc)
+	if !errors.Is(err, ErrUnavailable) || got.BookURL != "" {
+		t.Fatalf("expected unavailable search without a match, got %+v, %v", got, err)
+	}
+	stored, err := idx.DocumentByID(doc.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if mustData(t, stored).Enriched {
+		t.Fatal("failed search must remain pending")
+	}
+	if !slices.Equal(source.Queries(), []string{title + " " + author}) {
+		t.Fatalf("failed search must not trigger a title-only query, got %q", source.Queries())
+	}
+}
+
 func TestMatchesTolkienOmnibusSearchResults(t *testing.T) {
 	const title = "El señor de los anillos"
 	const wantURL = "/book/show/37910810-el-se-or-de-los-anillos"

internal/datasource/biblioreads/client.go

@@ -89,7 +89,8 @@ func WithCacheTTL(ttl time.Duration) Option {
 // instances ready to accept a request and randomizing ties. Requests to each
 // instance are spaced at least one second apart by default, including failover
 // attempts. Unavailable instances are skipped until their Retry-After time, or
-// httpjson.DefaultRetryAfter, has passed. Successful responses are cached in memory for
+// httpjson.DefaultRetryAfter, has passed. Book searches wait if all instances are cooling down.
+// Successful responses are cached in memory for
 // DefaultCacheTTL unless changed with WithCacheTTL. No endpoints are configured
 // by default; callers must supply them with WithInstances.
 func NewClient(opts ...Option) *Client {
@@ -131,7 +132,7 @@ type reviewsPayload struct {
 	LanguageCode  string `json:"languageCode,omitempty"`
 }
 
-// SearchBooks searches Goodreads books matching query.
+// SearchBooks searches Goodreads books matching query, with at most three attempts on retryable errors.
 func (c *Client) SearchBooks(ctx context.Context, query string) (SearchBooksResponse, error) {
 	var res SearchBooksResponse
 	payload := queryURLPayload{
@@ -212,8 +213,15 @@ func (c *Client) post(ctx context.Context, path string, payload, dest any) error
 	var errs []error
 	var incompleteResponse []byte
 	tried := make(map[string]bool)
-	for len(tried) < len(c.instances) {
-		instance, err := c.acquireInstance(ctx, tried)
+	attempts := len(c.instances)
+	if path == searchBooksPath {
+		attempts = 3
+	}
+	for range attempts {
+		if len(tried) == len(c.instances) {
+			clear(tried)
+		}
+		instance, err := c.acquireInstance(ctx, tried, path == searchBooksPath)
 		if err != nil {
 			if ctx.Err() != nil {
 				return ctx.Err()
@@ -310,27 +318,35 @@ func supportsReviewLanguage(version string) bool {
 
 // acquireInstance rechecks availability after waiting, so cancellation does not
 // reserve a future slot and concurrent requests cannot take the same slot.
-func (c *Client) acquireInstance(ctx context.Context, tried map[string]bool) (string, error) {
+func (c *Client) acquireInstance(ctx context.Context, tried map[string]bool, waitForCooldown bool) (string, error) {
 	for {
 		if err := ctx.Err(); err != nil {
 			return "", err
 		}
 		c.mu.Lock()
 		start := rand.IntN(len(c.instances))
 		instance := ""
+		var readyAt time.Time
 		now := time.Now()
 		for i := range c.instances {
 			candidate := c.instances[(start+i)%len(c.instances)]
-			if !tried[candidate] && !now.Before(c.cooldownUntil[candidate]) &&
-				(instance == "" || c.nextRequest[candidate].Before(c.nextRequest[instance])) {
+			if tried[candidate] || (!waitForCooldown && now.Before(c.cooldownUntil[candidate])) {
+				continue
+			}
+			ready := c.nextRequest[candidate]
+			if c.cooldownUntil[candidate].After(ready) {
+				ready = c.cooldownUntil[candidate]
+			}
+			if instance == "" || ready.Before(readyAt) {
 				instance = candidate
+				readyAt = ready
 			}
 		}
 		if instance == "" {
 			c.mu.Unlock()
 			return "", fmt.Errorf("%w: no untried instance is outside its cooldown", ErrUnavailable)
 		}
-		wait := time.Until(c.nextRequest[instance])
+		wait := time.Until(readyAt)
 		if wait <= 0 {
 			c.nextRequest[instance] = time.Now().Add(c.requestInterval)
 			c.mu.Unlock()
@@ -391,7 +407,14 @@ func (c *Client) postToInstance(ctx context.Context, endpoint string, body []byt
 	}
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("biblioreads: request to %s failed with status code %d: %s", endpoint, resp.StatusCode, resp.Body)
+		responseErr := fmt.Errorf("biblioreads: request to %s failed with status code %d: %s", endpoint, resp.StatusCode, resp.Body)
+		if resp.StatusCode == http.StatusNotFound && strings.HasSuffix(endpoint, searchBooksPath) {
+			return nil, &unavailableError{
+				endpoint: endpoint, retryAfter: httpjson.ParseRetryAfter(resp.Header.Get("Retry-After")),
+				reason: responseErr,
+			}
+		}
+		return nil, responseErr
 	}
 	return resp.Body, nil
 }

internal/datasource/biblioreads/client_test.go

@@ -4,12 +4,15 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"reflect"
+	"strings"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/svera/coreander/v5/internal/datasource/biblioreads"
@@ -393,6 +396,121 @@ func TestErrorStatus(t *testing.T) {
 	}
 }
 
+type searchTransport func(*http.Request) (*http.Response, error)
+
+func (transport searchTransport) RoundTrip(request *http.Request) (*http.Response, error) {
+	return transport(request)
+}
+
+func TestSearchRetries(t *testing.T) {
+	fixture := string(readFixture(t, "search-books.json"))
+	for _, tt := range []struct {
+		name        string
+		status      int
+		body        string
+		unavailable bool
+		attempts    int
+		recover     bool
+	}{
+		{"invalid query", http.StatusNotFound, `{"status":"Error - Invalid Query"}`, true, 3, false},
+		{"legitimate not found", http.StatusNotFound, `{"status":"Error - Book not found"}`, true, 3, false},
+		{"non-JSON not found", http.StatusNotFound, "<html>Not found</html>", true, 3, false},
+		{"other client error", http.StatusBadRequest, `{"status":"Error - Invalid Query"}`, false, 1, false},
+		{"server error", http.StatusServiceUnavailable, `{"status":"Error - Invalid Query"}`, true, 3, false},
+		{"rate limit", http.StatusTooManyRequests, `{}`, true, 3, false},
+		{"recovery", http.StatusNotFound, `{"status":"Error - Invalid Query"}`, false, 3, true},
+		{"empty success", http.StatusOK, `{"status":"Received","result":[]}`, false, 1, false},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				attempts := 0
+				client := newTestClient(biblioreads.WithInstances([]string{"https://example.test"}),
+					biblioreads.WithHTTPClient(&http.Client{Transport: searchTransport(func(r *http.Request) (*http.Response, error) {
+						var captured capturedRequest
+						if !captureRequest(t, httptest.NewRecorder(), r, &captured) {
+							t.Fatal("invalid search request")
+						}
+						assertRequest(t, captured, "/api/search/books", map[string]any{
+							"queryURL": "https://www.goodreads.com/search?q=El%20economista%20camuflado%20Tim%20Harford",
+						})
+						attempts++
+						status, body := tt.status, tt.body
+						if tt.recover && attempts == 3 {
+							status, body = http.StatusOK, fixture
+						}
+						header := make(http.Header)
+						if status == http.StatusTooManyRequests {
+							header.Set("Retry-After", "1")
+						}
+						return &http.Response{StatusCode: status, Header: header, Body: io.NopCloser(strings.NewReader(body))}, nil
+					})}))
+				start := time.Now()
+				response, err := client.SearchBooks(context.Background(), "El economista camuflado Tim Harford")
+				wantError := !tt.recover && tt.status != http.StatusOK
+				if (err != nil) != wantError || errors.Is(err, biblioreads.ErrUnavailable) != tt.unavailable {
+					t.Fatalf("unexpected search error classification: %v", err)
+				}
+				if attempts != tt.attempts {
+					t.Fatalf("expected %d attempts, got %d", tt.attempts, attempts)
+				}
+				cooldown := time.Minute
+				if tt.status == http.StatusTooManyRequests {
+					cooldown = time.Second
+				}
+				if attempts == 3 && time.Since(start) != 2*cooldown {
+					t.Fatalf("retries must respect the instance cooldown: %s", time.Since(start))
+				}
+				if tt.recover && len(response.Result) == 0 {
+					t.Fatal("successful retry must return its results")
+				}
+			})
+		})
+	}
+}
+
+func TestSearchRetryWaitHonorsCancellation(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		attempts := 0
+		client := newTestClient(biblioreads.WithInstances([]string{"https://example.test"}),
+			biblioreads.WithHTTPClient(&http.Client{Transport: searchTransport(func(*http.Request) (*http.Response, error) {
+				attempts++
+				return &http.Response{StatusCode: http.StatusNotFound, Header: make(http.Header),
+					Body: io.NopCloser(strings.NewReader(`{"status":"Error - Invalid Query"}`))}, nil
+			})}))
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		start := time.Now()
+		_, err := client.SearchBooks(ctx, "book")
+		if !errors.Is(err, context.DeadlineExceeded) || attempts != 1 || time.Since(start) != 10*time.Second {
+			t.Fatalf("cancellation must interrupt cooldown: attempts=%d, elapsed=%s, err=%v", attempts, time.Since(start), err)
+		}
+	})
+}
+
+func TestSearchInvalidQueryFailsOverWithoutChangingQuery(t *testing.T) {
+	var hits atomic.Int32
+	fixture := readFixture(t, "search-books.json")
+	client := newFailoverClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var captured capturedRequest
+		if !captureRequest(t, w, r, &captured) {
+			return
+		}
+		assertRequest(t, captured, "/api/search/books", map[string]any{
+			"queryURL": "https://www.goodreads.com/search?q=El%20economista%20camuflado%20Tim%20Harford",
+		})
+		if hits.Add(1) == 1 {
+			w.WriteHeader(http.StatusNotFound)
+			_, _ = w.Write([]byte(`{"status":"Error - Invalid Query","version":"BiblioReads v4.4.0"}`))
+			return
+		}
+		_, _ = w.Write(fixture)
+	}))
+	response, err := client.SearchBooks(context.Background(), "El economista camuflado Tim Harford")
+	if err != nil || len(response.Result) == 0 || hits.Load() != 2 {
+		t.Fatalf("expected same-query recovery on another instance, got %+v, %v, requests=%d", response, err, hits.Load())
+	}
+}
+
 func newCountingServer(t *testing.T, status int, body string, hits *atomic.Int32) *httptest.Server {
 	t.Helper()
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

internal/index/bleve.go

@@ -244,11 +244,13 @@ type progressTracker struct {
 	startNanos atomic.Int64
 	processed  atomic.Uint64
 	total      atomic.Uint64
+	failed     atomic.Bool
 }
 
 // begin marks the phase as started, resetting processed to 0 and total to
 // the given count.
 func (p *progressTracker) begin(total int) {
+	p.failed.Store(false)
 	p.startNanos.Store(time.Now().UnixNano())
 	p.processed.Store(0)
 	p.total.Store(uint64(total))

internal/index/bleve_document_enrich.go

@@ -76,13 +76,16 @@ func (b *BleveIndexer) documentEnrichTracker(i int) *progressTracker {
 	return &progressTracker{}
 }
 
-func (b *BleveIndexer) enrichProvider(provider DocumentEnricher, tracker *progressTracker, batchSize int) error {
+func (b *BleveIndexer) enrichProvider(provider DocumentEnricher, tracker *progressTracker, batchSize int) (resultErr error) {
+	defer func() {
+		tracker.failed.Store(resultErr != nil)
+		tracker.end()
+	}()
 	_, total, err := b.documentsNeedingProvider(provider.Source(), 0, nil)
 	if err != nil || total == 0 {
 		return err
 	}
 	tracker.begin(int(total))
-	defer tracker.end()
 	log.Printf("Enriching %d documents with %s data", total, provider.Source())
 	var failures []error
 	failedDocuments := make(map[string]struct{})

internal/index/bleve_document_enrich_test.go

@@ -49,6 +49,9 @@ func TestDocumentEnrichmentUsesBoundedConcurrency(t *testing.T) {
 	}
 	idx, _ := newTestTextRankIndexer(t, Config{})
 	documents := make([]Document, documentCount)
+	for i := range documents {
+		documents[i] = Document{ID: fmt.Sprintf("book-%d", i)}
+	}
 	done := make(chan map[string]error, 1)
 	go func() { done <- idx.enrichProviderBatch(provider, &progressTracker{}, documents) }()
 	defer func() {
@@ -126,8 +129,8 @@ func TestBackgroundProgressTracksPersistedProviderLookups(t *testing.T) {
 		t.Fatal("expected unavailable provider to fail")
 	}
 	progress, err = idx.BackgroundProgress()
-	if err != nil || len(progress) != 1 || progress[0].Name != "first" || progress[0].Percentage != 66 {
-		t.Fatalf("failed lookups must remain visible; successful providers must disappear: %+v, %v", progress, err)
+	if err != nil || len(progress) != 0 {
+		t.Fatalf("finished providers must disappear, including failed lookups: %+v, %v", progress, err)
 	}
 	// Progress is reconstructed from stored data, not a running worker counter.
 	resumed := &BleveIndexer{documentsIdx: idx.documentsIdx, documentEnrichers: []DocumentEnricher{first}}
@@ -228,6 +231,18 @@ func TestDocumentEnrichmentContinuesAfterDocumentFailure(t *testing.T) {
 	if len(pending) != 1 || pending[0].ID != "book-2" {
 		t.Fatalf("failed document should remain pending for a later retry: %+v", pending)
 	}
+	progress, err := idx.BackgroundProgress()
+	if err != nil || len(progress) != 0 {
+		t.Fatalf("failed run must not report active progress: %+v, %v", progress, err)
+	}
+	provider.failure = nil
+	if err := idx.EnrichDocuments(1); err != nil {
+		t.Fatal(err)
+	}
+	progress, err = idx.BackgroundProgress()
+	if err != nil || len(progress) != 0 || idx.documentEnrichProgress[0].failed.Load() {
+		t.Fatalf("successful retry must clear the failed state: %+v, %v", progress, err)
+	}
 }
 
 func TestDocumentEnrichmentSupportsIndependentProviders(t *testing.T) {

internal/index/bleve_document_write.go

@@ -81,11 +81,15 @@ func (b *BleveIndexer) BackgroundProgress() ([]Progress, error) {
 	}
 	for i, counts := range providers {
 		if counts.enriched < counts.total {
+			tracker := b.documentEnrichTracker(i)
+			if tracker.failed.Load() {
+				continue
+			}
 			progress = append(progress, Progress{
 				Kind: ProgressDataSources, InProgress: true, Name: b.documentEnrichers[i].Name(),
 				// Do not round pending work up to 100%.
 				Percentage:    math.Floor(100 * float64(counts.enriched) / float64(counts.total)),
-				RemainingTime: b.documentEnrichTracker(i).remainingTime(counts.total - counts.enriched),
+				RemainingTime: tracker.remainingTime(counts.total - counts.enriched),
 			})
 		}
 	}