pdp browser rendering

Author: LexLuthr

lib/cachedreader/cachedreader.go

@@ -81,9 +81,10 @@ type cachedSectionReader struct {
 	// err is non-nil if there's an error getting the underlying piece reader
 	err error
 	// cancel for underlying GetPieceReader call
-	cancel  func()
-	refs    int
-	expired bool
+	cancel   func()
+	refs     int
+	expired  bool
+	pdpPiece bool
 }
 
 func (r *cachedSectionReader) Close() error {
@@ -192,7 +193,7 @@ func (cpr *CachedPieceReader) getPieceReaderFromPiecePark(ctx context.Context, p
 	return reader, abi.UnpaddedPieceSize(pieceData[0].PieceRawSize), nil
 }
 
-func (cpr *CachedPieceReader) GetSharedPieceReader(ctx context.Context, pieceCid cid.Cid) (storiface.Reader, abi.UnpaddedPieceSize, error) {
+func (cpr *CachedPieceReader) GetSharedPieceReader(ctx context.Context, pieceCid cid.Cid) (storiface.Reader, abi.UnpaddedPieceSize, bool, error) {
 	var r *cachedSectionReader
 
 	// Check if there is already a piece reader in the cache
@@ -228,8 +229,9 @@ func (cpr *CachedPieceReader) GetSharedPieceReader(ctx context.Context, pieceCid
 				r.err = fmt.Errorf("failed to get piece reader from sector or piece park: %w, %w", err, serr)
 				readerCtxCancel()
 
-				return nil, 0, r.err
+				return nil, 0, false, r.err
 			}
+			r.pdpPiece = true
 		}
 
 		r.reader = reader
@@ -248,7 +250,7 @@ func (cpr *CachedPieceReader) GetSharedPieceReader(ctx context.Context, pieceCid
 			// The context timed out. Dereference the cached piece reader and
 			// return an error.
 			_ = r.Close()
-			return nil, 0, ctx.Err()
+			return nil, 0, false, ctx.Err()
 		case <-r.ready:
 		}
 	}
@@ -257,7 +259,7 @@ func (cpr *CachedPieceReader) GetSharedPieceReader(ctx context.Context, pieceCid
 	// that the cached reader gets cleaned up
 	if r.err != nil {
 		_ = r.Close()
-		return nil, 0, r.err
+		return nil, 0, false, r.err
 	}
 
 	rs := io.NewSectionReader(r.reader, 0, int64(r.pieceSize))
@@ -272,5 +274,5 @@ func (cpr *CachedPieceReader) GetSharedPieceReader(ctx context.Context, pieceCid
 		Reader:   rs,
 		Seeker:   rs,
 		ReaderAt: r.reader,
-	}, r.pieceSize, nil
+	}, r.pieceSize, r.pdpPiece, nil
 }

market/ipni/chunker/serve-chunker.go

@@ -218,7 +218,7 @@ func (p *ServeChunker) getEntry(rctx context.Context, block cid.Cid, speculated
 func (p *ServeChunker) reconstructChunkFromCar(ctx context.Context, chunk, piece cid.Cid, startOff int64, next ipld.Link, numBlocks int64, speculate bool) ([]byte, error) {
 	start := time.Now()
 
-	reader, _, err := p.cpr.GetSharedPieceReader(ctx, piece)
+	reader, _, _, err := p.cpr.GetSharedPieceReader(ctx, piece)
 	defer func(reader storiface.Reader) {
 		_ = reader.Close()
 	}(reader)

market/retrieval/piecehandler.go

@@ -45,7 +45,7 @@ func (rp *Provider) handleByPieceCid(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Get a reader over the piece
-	reader, size, err := rp.cpr.GetSharedPieceReader(ctx, pieceCid)
+	reader, size, pdpPiece, err := rp.cpr.GetSharedPieceReader(ctx, pieceCid)
 	if err != nil {
 		log.Errorf("server error getting content for piece CID %s: %s", pieceCid, err)
 		if errors.Is(err, cachedreader.NoDealErr) {
@@ -58,19 +58,22 @@ func (rp *Provider) handleByPieceCid(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	setHeaders(w, pieceCid)
+	setHeaders(w, pieceCid, pdpPiece)
 	serveContent(w, r, size, reader)
 
 	stats.Record(ctx, remoteblockstore.HttpPieceByCid200ResponseCount.M(1))
 	stats.Record(ctx, remoteblockstore.HttpPieceByCidRequestDuration.M(float64(time.Since(startTime).Milliseconds())))
 }
 
-func setHeaders(w http.ResponseWriter, pieceCid cid.Cid) {
+func setHeaders(w http.ResponseWriter, pieceCid cid.Cid, pdpPiece bool) {
 	w.Header().Set("Vary", "Accept-Encoding")
 	etag := `"` + pieceCid.String() + `.gz"` // must be quoted
 	w.Header().Set("Etag", etag)
 	w.Header().Set("Content-Type", "application/piece")
 	w.Header().Set("Cache-Control", "public, max-age=29030400, immutable")
+	if pdpPiece {
+		w.Header().Set("Content-Disposition", "inline")
+	}
 }
 
 func serveContent(res http.ResponseWriter, req *http.Request, size abi.UnpaddedPieceSize, content io.ReadSeeker) {

market/retrieval/remoteblockstore/remoteblockstore.go

@@ -116,7 +116,7 @@ func (ro *RemoteBlockstore) Get(ctx context.Context, c cid.Cid) (b blocks.Block,
 	for _, piece := range pieces {
 		data, err := func() ([]byte, error) {
 			// Get a reader over the piece data
-			reader, _, err := ro.cpr.GetSharedPieceReader(ctx, piece.PieceCid)
+			reader, _, _, err := ro.cpr.GetSharedPieceReader(ctx, piece.PieceCid)
 			if err != nil {
 				return nil, fmt.Errorf("getting piece reader: %w", err)
 			}

tasks/pdp/task_prove.go

@@ -392,11 +392,15 @@ func (p *ProveTask) genSubrootMemtree(ctx context.Context, subrootCid string, su
 		return nil, xerrors.Errorf("subroot size exceeds maximum: %d", subrootSize)
 	}
 
-	subrootReader, unssize, err := p.cpr.GetSharedPieceReader(ctx, subrootCidObj)
+	subrootReader, unssize, pdpPiece, err := p.cpr.GetSharedPieceReader(ctx, subrootCidObj)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to get subroot reader: %w", err)
 	}
 
+	if !pdpPiece {
+		return nil, xerrors.Errorf("subroot is not a PDP piece")
+	}
+
 	var r io.Reader = subrootReader
 
 	if unssize.Padded() > subrootSize {