Read a large library: page the section read, and chunk the show walk

docs/troubleshooting.md

@@ -22,6 +22,51 @@ not submit anything, so there is no `submit` command. If you want to contribute
 timing you know, do it on [theintrodb.org](https://theintrodb.org), which is
 where contributions are made.
 
+## "read the Plex library: ... timeout or cancel"
+
+The request that reads a library section did not come back inside
+`plex.timeout_s` (20 seconds by default).
+
+The library read is paged, the first request included, so no single request asks
+Plex to build a response proportional to the size of the section. On a build
+older than that fix the first request was unpaged, and Plex answered it by
+assembling the whole section at once — which on a section of tens of thousands
+of episodes outlives any sensible timeout. Raising `plex.timeout_s` was the
+workaround then:
+
+```toml
+[plex]
+timeout_s = 120
+```
+
+It still works as a workaround, but it is no longer the fix, and it is worth
+checking that the failing request is the section read before reaching for it: a
+timeout on `/library/metadata/<key>` is a different problem.
+
+`/identity` answering — which is what `setup` and `config check` report as `plex
+server ok` — proves only that the URL reaches a Plex server. It is
+unauthenticated, so it says nothing about the token, and it is a small answer, so
+it says nothing about whether a library-sized one can be served. A rejected token
+answers 401 immediately rather than timing out.
+
+## "could not read the shows behind the episodes; they keep their own ids"
+
+The run could not read the show behind each episode out of the Plex database, so
+every episode keeps the provider id in its own row. That id is the episode's, and
+TheIntroDB is asked for an episode by series id, so those lookups answer "media
+not found" — the run spends its allowance on questions that cannot succeed, and
+no TV markers are written. It is one WARN line and no other symptom, which is why
+it is worth reading.
+
+The known cause was a library larger than SQLite's bound-parameter limit: the
+walk bound one parameter per episode in a single statement, and above 32,766
+(`SQLITE_MAX_VARIABLE_NUMBER`) the statement failed outright with `too many SQL
+variables`. The walk is chunked now, so a library of any size reads.
+
+The other causes are ordinary: no read access to the database, or episodes whose
+show rows carry no provider ids. Check what `plex-sync library` prints for the
+ids an item would be looked up with.
+
 ## "no-provider-id"
 
 The item has no TMDb, IMDb or Tvdb id, so there is nothing to look it up by.

internal/plexapi/client.go

@@ -242,23 +242,25 @@ func (c *Client) Items(ctx context.Context, sectionKeys []int) ([]model.LibraryI
 	return out, nil
 }
 
-// sectionItems reads one section, following Plex's container paging when the
-// server reports more items than the first answer carried.
+// sectionItems reads one section, a page at a time.
+//
+// Every request carries the container headers, the first one included. Plex
+// answers an unpaged /all with the whole section, and on a library of tens of
+// thousands of episodes that single answer is the request that never finishes:
+// the server is still assembling it when the client's timeout expires, which
+// surfaces as "timeout or cancel" on a library that is perfectly healthy. The
+// first request is therefore bounded like every other one, so nothing ever asks
+// Plex to build a response proportional to the library.
 func (c *Client) sectionItems(ctx context.Context, key, metadataType int, kind model.Kind) ([]model.LibraryItem, error) {
 	path := "/library/sections/" + strconv.Itoa(key) + "/all"
 	query := url.Values{
 		"type":         {strconv.Itoa(metadataType)},
 		"includeGuids": {"1"},
 	}
 
-	ctr, err := c.container(ctx, path, query, nil)
-	if err != nil {
-		return nil, err
-	}
-	out := parseItems(*ctr, kind)
-	total := ctr.MediaContainer.Total.Int()
-
-	for start := len(out); total > 0 && start < total; {
+	var out []model.LibraryItem
+	seen := make(map[int]bool)
+	for start := 0; ; {
 		extra := map[string]string{
 			"X-Plex-Container-Start": strconv.Itoa(start),
 			"X-Plex-Container-Size":  strconv.Itoa(ItemWindow),
@@ -268,11 +270,34 @@ func (c *Client) sectionItems(ctx context.Context, key, metadataType int, kind m
 			return out, err
 		}
 		batch := parseItems(*page, kind)
-		if len(batch) == 0 {
+
+		added := 0
+		for _, it := range batch {
+			if seen[it.RatingKey] {
+				continue
+			}
+			seen[it.RatingKey] = true
+			out = append(out, it)
+			added++
+		}
+		// A page with nothing new on it is a server repeating itself, which is
+		// what a server that ignores the offset would do forever. It is also
+		// what the walk ends on when the server reports no total: every
+		// iteration either adds an item or stops, and the library is finite, so
+		// the walk terminates without a page size to count against.
+		if added == 0 {
 			break
 		}
-		out = append(out, batch...)
 		start += len(batch)
+
+		// When the server reports a total, it is the only thing that says the
+		// walk is over. The page size deliberately is not: a server that caps
+		// its answer below the size asked for would otherwise end the walk on
+		// its first page and silently return a fraction of the library, which
+		// is worse than the timeout this paging exists to avoid.
+		if total := page.MediaContainer.Total.Int(); total > 0 && start >= total {
+			break
+		}
 	}
 	return out, nil
 }

internal/plexapi/client_test.go

@@ -527,6 +527,197 @@ func TestItemsPaging(t *testing.T) {
 	}
 }
 
+// TestItemsFirstRequestIsPaged pins the fix for a library that times out before
+// it is read: an unpaged /all makes Plex assemble the whole section, and on a
+// section of tens of thousands of episodes that answer outlives the client's
+// timeout. The fake below behaves the way a real server does — it answers an
+// unpaged request slowly and a paged one immediately — so the old code fails
+// here with "timeout or cancel" and this one does not.
+func TestItemsFirstRequestIsPaged(t *testing.T) {
+	t.Parallel()
+	const total = 500
+
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/library/sections/7/all" {
+			writeJSON(w, http.StatusOK,
+				`{"MediaContainer":{"size":1,"Directory":[{"key":"7","title":"TV","type":"show"}]}}`)
+			return
+		}
+		raw := r.Header.Get("X-Plex-Container-Size")
+		if raw == "" {
+			// Unpaged: Plex builds the entire section, which is the request
+			// that never returns in time on a large library.
+			time.Sleep(2 * time.Second)
+			writeJSON(w, http.StatusOK, sectionPage(0, total, total))
+			return
+		}
+		start, _ := strconv.Atoi(r.Header.Get("X-Plex-Container-Start"))
+		size, _ := strconv.Atoi(raw)
+		if size <= 0 {
+			size = ItemWindow
+		}
+		writeJSON(w, http.StatusOK, sectionPage(start, size, total))
+	}
+
+	f := newFake(t, handler)
+	hc := httpclient.New(300*time.Millisecond, false, UserAgent)
+	t.Cleanup(hc.Close)
+	c := NewClient(config.Plex{URL: f.srv.URL, Token: fakeToken}, hc)
+
+	items, err := c.Items(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("Items: %v (the first request was not paged)", err)
+	}
+	if len(items) != total {
+		t.Fatalf("Items = %d entries, want %d", len(items), total)
+	}
+	for i, it := range items {
+		if it.RatingKey != 900+i {
+			t.Fatalf("items[%d].RatingKey = %d, want %d", i, it.RatingKey, 900+i)
+		}
+	}
+}
+
+// TestItemsPagingWithoutTotal covers a server that reports no total, which is
+// the one case where the page size is the only thing that says the walk is over.
+func TestItemsPagingWithoutTotal(t *testing.T) {
+	t.Parallel()
+	const total = ItemWindow + 7
+
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/library/sections/7/all" {
+			writeJSON(w, http.StatusOK,
+				`{"MediaContainer":{"size":1,"Directory":[{"key":"7","title":"TV","type":"show"}]}}`)
+			return
+		}
+		start, _ := strconv.Atoi(r.Header.Get("X-Plex-Container-Start"))
+		size, _ := strconv.Atoi(r.Header.Get("X-Plex-Container-Size"))
+		if size <= 0 {
+			size = ItemWindow
+		}
+		writeJSON(w, http.StatusOK, sectionPageNoTotal(start, size, total))
+	}
+
+	f := newFake(t, handler)
+	c := f.client(t)
+
+	items, err := c.Items(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("Items: %v", err)
+	}
+	if len(items) != total {
+		t.Fatalf("Items = %d entries, want %d", len(items), total)
+	}
+}
+
+// TestItemsPagingServerThatCapsThePage covers a server that answers with fewer
+// items than the size asked for and reports no total. The page size must not end
+// the walk: doing so would return a fraction of the library and look like a
+// successful read, which is worse than the timeout the paging exists to avoid.
+func TestItemsPagingServerThatCapsThePage(t *testing.T) {
+	t.Parallel()
+	const (
+		total  = 7
+		server = 2 // what this server will hand over per request
+	)
+
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/library/sections/7/all" {
+			writeJSON(w, http.StatusOK,
+				`{"MediaContainer":{"size":1,"Directory":[{"key":"7","title":"TV","type":"show"}]}}`)
+			return
+		}
+		start, _ := strconv.Atoi(r.Header.Get("X-Plex-Container-Start"))
+		writeJSON(w, http.StatusOK, sectionPageNoTotal(start, server, total))
+	}
+
+	f := newFake(t, handler)
+	c := f.client(t)
+
+	items, err := c.Items(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("Items: %v", err)
+	}
+	if len(items) != total {
+		t.Fatalf("Items = %d entries, want %d: a capped page ended the walk early", len(items), total)
+	}
+	for i, it := range items {
+		if it.RatingKey != 900+i {
+			t.Fatalf("items[%d].RatingKey = %d, want %d", i, it.RatingKey, 900+i)
+		}
+	}
+}
+
+// TestItemsServerThatIgnoresPaging terminates and does not duplicate when the
+// server answers every request with the whole section, which is what a server
+// with no container support looks like.
+func TestItemsServerThatIgnoresPaging(t *testing.T) {
+	t.Parallel()
+	const total = 3
+
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/library/sections/7/all" {
+			writeJSON(w, http.StatusOK,
+				`{"MediaContainer":{"size":1,"Directory":[{"key":"7","title":"TV","type":"show"}]}}`)
+			return
+		}
+		writeJSON(w, http.StatusOK, sectionPage(0, total, total))
+	}
+
+	f := newFake(t, handler)
+	c := f.client(t)
+
+	items, err := c.Items(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("Items: %v", err)
+	}
+	if len(items) != total {
+		t.Fatalf("Items = %d entries, want %d with no duplicates", len(items), total)
+	}
+}
+
+// sectionPage renders a page of a section that reports its total.
+func sectionPage(start, size, total int) string {
+	if start < 0 || start > total {
+		start = total
+	}
+	end := start + size
+	if end > total {
+		end = total
+	}
+	return page(start, end, total, true)
+}
+
+// sectionPageNoTotal renders a page from a server that reports no total.
+func sectionPageNoTotal(start, size, total int) string {
+	if start < 0 || start > total {
+		start = total
+	}
+	end := start + size
+	if end > total {
+		end = total
+	}
+	return page(start, end, total, false)
+}
+
+func page(start, end, total int, withTotal bool) string {
+	var sb strings.Builder
+	sb.WriteString(`{"MediaContainer":{"size":`)
+	fmt.Fprintf(&sb, "%d", end-start)
+	if withTotal {
+		fmt.Fprintf(&sb, `,"total":%d`, total)
+	}
+	sb.WriteString(`,"Metadata":[`)
+	for i := start; i < end; i++ {
+		if i > start {
+			sb.WriteString(",")
+		}
+		fmt.Fprintf(&sb, `{"ratingKey":%d,"type":"movie","title":"Movie %d"}`, 900+i, i)
+	}
+	sb.WriteString(`]}}`)
+	return sb.String()
+}
+
 func TestChapters(t *testing.T) {
 	t.Parallel()
 	f := newFake(t, plexHandler)