Monorepo for Tangled
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>
oppiliappan
(Jul 9, 2026, 2:01 PM +0100)
ae3b7cd4
c661a193
+14
-27
+9
-20
appview/ingester.go
+9
-20
appview/ingester.go
···
1
1
package appview
2
2
3
3
import (
4
+
"bytes"
4
5
"context"
5
6
"database/sql"
6
7
"encoding/json"
···
12
13
"net/url"
13
14
"slices"
14
15
"strings"
15
-
"sync"
16
16
17
17
"time"
18
18
···
1466
1466
}
1467
1467
1468
1468
// go through and fetch all blobs in parallel
1469
-
readers := make([]*io.ReadCloser, len(record.Rounds))
1470
-
var mu sync.Mutex
1469
+
blobs := make([]io.Reader, len(record.Rounds))
1471
1470
1472
1471
g, gctx := errgroup.WithContext(ctx)
1473
1472
···
1497
1496
l.Error("failed to make request")
1498
1497
return err
1499
1498
}
1499
+
defer resp.Body.Close()
1500
1500
1501
-
mu.Lock()
1502
-
readers[idx] = &resp.Body
1503
-
mu.Unlock()
1501
+
var buf bytes.Buffer
1502
+
if _, err := io.Copy(&buf, io.LimitReader(resp.Body, 16<<20)); err != nil {
1503
+
return fmt.Errorf("failed to read blob in round %d: %w", idx, err)
1504
+
}
1505
+
blobs[idx] = &buf
1504
1506
1505
1507
return nil
1506
1508
})
1507
1509
}
1508
1510
1509
1511
if err := g.Wait(); err != nil {
1510
-
for _, r := range readers {
1511
-
if r != nil && *r != nil {
1512
-
(*r).Close()
1513
-
}
1514
-
}
1515
1512
return err
1516
1513
}
1517
1514
1518
-
defer func() {
1519
-
for _, r := range readers {
1520
-
if r != nil && *r != nil {
1521
-
(*r).Close()
1522
-
}
1523
-
}
1524
-
}()
1525
-
1526
-
pull, err := models.PullFromRecord(did, rkey, record, readers)
1515
+
pull, err := models.PullFromRecord(did, rkey, record, blobs)
1527
1516
if err != nil {
1528
1517
return fmt.Errorf("failed to parse pull from record: %w", err)
1529
1518
}
+4
-4
appview/models/pull.go
+4
-4
appview/models/pull.go
···
158
158
return nil
159
159
}
160
160
161
-
func PullFromRecord(did, rkey string, record tangled.RepoPull, blobs []*io.ReadCloser) (*Pull, error) {
161
+
func PullFromRecord(did, rkey string, record tangled.RepoPull, blobs []io.Reader) (*Pull, error) {
162
162
created, err := time.Parse(time.RFC3339, record.CreatedAt)
163
163
if err != nil {
164
164
return nil, fmt.Errorf("invalid createdAt: %w", err)
···
213
213
214
214
var submissions []*PullSubmission
215
215
for i, s := range record.Rounds {
216
-
var blob *io.ReadCloser
216
+
var blob io.Reader
217
217
if i < len(blobs) {
218
218
blob = blobs[i]
219
219
}
···
239
239
}, nil
240
240
}
241
241
242
-
func PullSubmissionFromRecord(did, rkey string, roundNumber int, round *tangled.RepoPull_Round, blob *io.ReadCloser) (*PullSubmission, error) {
242
+
func PullSubmissionFromRecord(did, rkey string, roundNumber int, round *tangled.RepoPull_Round, blob io.Reader) (*PullSubmission, error) {
243
243
created, err := time.Parse(time.RFC3339, round.CreatedAt)
244
244
if err != nil {
245
245
return nil, fmt.Errorf("invalid createdAt: %w", err)
···
247
247
248
248
var patch, sourceRev string
249
249
if blob != nil {
250
-
p, err := extractGzip(*blob)
250
+
p, err := extractGzip(blob)
251
251
if err != nil {
252
252
return nil, fmt.Errorf("failed to extract gzip: %w", err)
253
253
}
+1
-3
spindle/tapclient.go
+1
-3
spindle/tapclient.go
···
6
6
"encoding/json"
7
7
"errors"
8
8
"fmt"
9
-
"io"
10
9
"log/slog"
11
10
"net/http"
12
11
"net/url"
···
554
553
}
555
554
defer blobResp.Body.Close()
556
555
557
-
blob := io.ReadCloser(blobResp.Body)
558
-
latestSubmission, err := avmodels.PullSubmissionFromRecord(did, rkey, roundNumber, round, &blob)
556
+
latestSubmission, err := avmodels.PullSubmissionFromRecord(did, rkey, roundNumber, round, blobResp.Body)
559
557
if err != nil {
560
558
return nil, fmt.Errorf("failed to parse submission: %w", err)
561
559
}