From 7da92f6640090336838ee7416b5b1caca715e968 Mon Sep 17 00:00:00 2001
From: oppiliappan <me@oppi.li>
Date: Tue, 7 Jul 2026 19:43:51 +0100
Subject: [PATCH] appview/ingester: fix pull ingestion

for some blobs, pull ingestion would spuriously fail. this was because
the surrounding context was cancelled as soon as gctx.Wait returned. for
smaller blobs, this did not affect blob fetches.

as a scientific test, some cleverly insert time.Sleeps can help
reproduce the issue for any blob size and on any setup (localinfra or
otherwise). this change permits blob contents to be read even when the
surrounding context is cancelled.

Signed-off-by: oppiliappan <me@oppi.li>
---
 appview/ingester.go    | 29 +++++++++--------------------
 appview/models/pull.go |  8 ++++----
 spindle/tapclient.go   |  4 +---
 3 files changed, 14 insertions(+), 27 deletions(-)

diff --git a/appview/ingester.go b/appview/ingester.go
index 8c0e9f29..13b60f95 100644
--- a/appview/ingester.go
+++ b/appview/ingester.go
@@ -1,6 +1,7 @@
 package appview
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
 	"encoding/json"
@@ -12,7 +13,6 @@ import (
 	"net/url"
 	"slices"
 	"strings"
-	"sync"
 
 	"time"
 
@@ -1474,8 +1474,7 @@ func (i *Ingester) ingestPull(ctx context.Context, e *jmodels.Event, l *slog.Log
 		}
 
 		// go through and fetch all blobs in parallel
-		readers := make([]*io.ReadCloser, len(record.Rounds))
-		var mu sync.Mutex
+		blobs := make([]io.Reader, len(record.Rounds))
 
 		g, gctx := errgroup.WithContext(ctx)
 
@@ -1505,33 +1504,23 @@ func (i *Ingester) ingestPull(ctx context.Context, e *jmodels.Event, l *slog.Log
 					l.Error("failed to make request")
 					return err
 				}
+				defer resp.Body.Close()
 
-				mu.Lock()
-				readers[idx] = &resp.Body
-				mu.Unlock()
+				var buf bytes.Buffer
+				if _, err := io.Copy(&buf, io.LimitReader(resp.Body, 16<<20)); err != nil {
+					return fmt.Errorf("failed to read blob in round %d: %w", idx, err)
+				}
+				blobs[idx] = &buf
 
 				return nil
 			})
 		}
 
 		if err := g.Wait(); err != nil {
-			for _, r := range readers {
-				if r != nil && *r != nil {
-					(*r).Close()
-				}
-			}
 			return err
 		}
 
-		defer func() {
-			for _, r := range readers {
-				if r != nil && *r != nil {
-					(*r).Close()
-				}
-			}
-		}()
-
-		pull, err := models.PullFromRecord(did, rkey, record, readers)
+		pull, err := models.PullFromRecord(did, rkey, record, blobs)
 		if err != nil {
 			return fmt.Errorf("failed to parse pull from record: %w", err)
 		}
diff --git a/appview/models/pull.go b/appview/models/pull.go
index e35d4d7f..3988b0d0 100644
--- a/appview/models/pull.go
+++ b/appview/models/pull.go
@@ -158,7 +158,7 @@ func (pull *Pull) Validate() error {
 	return nil
 }
 
-func PullFromRecord(did, rkey string, record tangled.RepoPull, blobs []*io.ReadCloser) (*Pull, error) {
+func PullFromRecord(did, rkey string, record tangled.RepoPull, blobs []io.Reader) (*Pull, error) {
 	created, err := time.Parse(time.RFC3339, record.CreatedAt)
 	if err != nil {
 		return nil, fmt.Errorf("invalid createdAt: %w", err)
@@ -213,7 +213,7 @@ func PullFromRecord(did, rkey string, record tangled.RepoPull, blobs []*io.ReadC
 
 	var submissions []*PullSubmission
 	for i, s := range record.Rounds {
-		var blob *io.ReadCloser
+		var blob io.Reader
 		if i < len(blobs) {
 			blob = blobs[i]
 		}
@@ -239,7 +239,7 @@ func PullFromRecord(did, rkey string, record tangled.RepoPull, blobs []*io.ReadC
 	}, nil
 }
 
-func PullSubmissionFromRecord(did, rkey string, roundNumber int, round *tangled.RepoPull_Round, blob *io.ReadCloser) (*PullSubmission, error) {
+func PullSubmissionFromRecord(did, rkey string, roundNumber int, round *tangled.RepoPull_Round, blob io.Reader) (*PullSubmission, error) {
 	created, err := time.Parse(time.RFC3339, round.CreatedAt)
 	if err != nil {
 		return nil, fmt.Errorf("invalid createdAt: %w", err)
@@ -247,7 +247,7 @@ func PullSubmissionFromRecord(did, rkey string, roundNumber int, round *tangled.
 
 	var patch, sourceRev string
 	if blob != nil {
-		p, err := extractGzip(*blob)
+		p, err := extractGzip(blob)
 		if err != nil {
 			return nil, fmt.Errorf("failed to extract gzip: %w", err)
 		}
diff --git a/spindle/tapclient.go b/spindle/tapclient.go
index 6060d2ca..e8683486 100644
--- a/spindle/tapclient.go
+++ b/spindle/tapclient.go
@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"log/slog"
 	"net/http"
 	"net/url"
@@ -554,8 +553,7 @@ func (t *Tap) fetchLatestSubmission(ctx context.Context, did, rkey string, recor
 	}
 	defer blobResp.Body.Close()
 
-	blob := io.ReadCloser(blobResp.Body)
-	latestSubmission, err := avmodels.PullSubmissionFromRecord(did, rkey, roundNumber, round, &blob)
+	latestSubmission, err := avmodels.PullSubmissionFromRecord(did, rkey, roundNumber, round, blobResp.Body)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse submission: %w", err)
 	}
-- 
2.51.2