beathagenlocher.com / core
Monorepo for Tangled
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

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>

authored by

oppiliappan and committed by
Tangled (Jul 10, 2026, 2:10 PM +0300) 7da92f66 ae6436b9

+14 -27
+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 ··· 1474 1474 } 1475 1475 1476 1476 // go through and fetch all blobs in parallel 1477 - readers := make([]*io.ReadCloser, len(record.Rounds)) 1478 - var mu sync.Mutex 1477 + blobs := make([]io.Reader, len(record.Rounds)) 1479 1478 1480 1479 g, gctx := errgroup.WithContext(ctx) 1481 1480 ··· 1505 1504 l.Error("failed to make request") 1506 1505 return err 1507 1506 } 1507 + defer resp.Body.Close() 1508 1508 1509 - mu.Lock() 1510 - readers[idx] = &resp.Body 1511 - mu.Unlock() 1509 + var buf bytes.Buffer 1510 + if _, err := io.Copy(&buf, io.LimitReader(resp.Body, 16<<20)); err != nil { 1511 + return fmt.Errorf("failed to read blob in round %d: %w", idx, err) 1512 + } 1513 + blobs[idx] = &buf 1512 1514 1513 1515 return nil 1514 1516 }) 1515 1517 } 1516 1518 1517 1519 if err := g.Wait(); err != nil { 1518 - for _, r := range readers { 1519 - if r != nil && *r != nil { 1520 - (*r).Close() 1521 - } 1522 - } 1523 1520 return err 1524 1521 } 1525 1522 1526 - defer func() { 1527 - for _, r := range readers { 1528 - if r != nil && *r != nil { 1529 - (*r).Close() 1530 - } 1531 - } 1532 - }() 1533 - 1534 - pull, err := models.PullFromRecord(did, rkey, record, readers) 1523 + pull, err := models.PullFromRecord(did, rkey, record, blobs) 1535 1524 if err != nil { 1536 1525 return fmt.Errorf("failed to parse pull from record: %w", err) 1537 1526 }
+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
··· 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 }