fix: Chunk Lens blocks if larger than 3MB (#145)

Author: AndrewSisley

host-go/node/node.go

@@ -69,6 +69,10 @@ func New(ctx context.Context, opts ...Option) (*Node, error) {
 		o.IndexstoreNamespace = immutable.Some("i")
 	}
 
+	if !o.MaxBlockSize.HasValue() {
+		o.MaxBlockSize = immutable.Some(1024 * 1024 * 3)
+	}
+
 	repo := repository.NewRepository(o.PoolSize.Value(), o.Runtime.Value(), &repositoryTxnSource{src: o.TxnSource.Value()})
 
 	node, err := createNode(
@@ -78,6 +82,7 @@ func New(ctx context.Context, opts ...Option) (*Node, error) {
 			repo,
 			o.BlockstoreNamespace.Value(),
 			o.BlockstoreChunkSize,
+			o.MaxBlockSize,
 			o.IndexstoreNamespace.Value(),
 		),
 		repo,

host-go/node/node_p2p.go

@@ -27,6 +27,7 @@ type Options struct {
 	Runtime             immutable.Option[module.Runtime]
 	BlockstoreNamespace immutable.Option[string]
 	BlockstoreChunkSize immutable.Option[int]
+	MaxBlockSize        immutable.Option[int]
 	IndexstoreNamespace immutable.Option[string]
 	P2P                 immutable.Option[p2p.Host]
 	DisableP2P          bool

host-go/node/option.go

@@ -70,3 +70,15 @@ func WithBlockstoreChunkSize(blockstoreChunkSize int) Option {
 		opts.BlockstoreChunkSize = immutable.Some(blockstoreChunkSize)
 	}
 }
+
+// WithMaxBlockSize will limit the size of lens blocks to the given length if provided.
+//
+// IPFS implmentations limit the size of the blocks that they are willing to transport
+// and can silently fail if they come across a block that is too large (~4MB).
+//
+// This option defaults to 3MB.
+func WithMaxBlockSize(maxBlockSize int) Option {
+	return func(opts *Options) {
+		opts.MaxBlockSize = immutable.Some(maxBlockSize)
+	}
+}

host-go/store/block.go

@@ -5,8 +5,10 @@
 package store
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
+	"slices"
 	"sort"
 
 	cid "github.com/ipfs/go-cid"
@@ -46,21 +48,26 @@ func init() {
 	LensBlockSchema, LensBlockSchemaPrototype = mustSetSchema(
 		"lensBlock",
 		&LensBlock{},
+		&Chunks{},
 	)
 }
 
-func mustSetSchema(schemaName string, schema schemaDefinition) (schema.Type, ipld.NodePrototype) {
-	ts, err := ipld.LoadSchemaBytes(schema.IPLDSchemaBytes())
+func mustSetSchema(schemaName string, schemas ...schemaDefinition) (schema.Type, ipld.NodePrototype) {
+	schemaBytes := make([][]byte, 0, len(schemas))
+	for _, s := range schemas {
+		schemaBytes = append(schemaBytes, s.IPLDSchemaBytes())
+	}
+
+	ts, err := ipld.LoadSchemaBytes(bytes.Join(schemaBytes, nil))
 	if err != nil {
 		panic(err)
 	}
-
 	schemaType := ts.TypeByName(schemaName)
 
 	// Calling bindnode.Prototype here ensure that [Block] and all the types it contains
 	// are compatible with the IPLD schema defined by [schemaDefinition].
 	// If [Block] and [schemaType] do not match, this will panic.
-	proto := bindnode.Prototype(schema, schemaType)
+	proto := bindnode.Prototype(schemas[0], schemaType)
 
 	return schemaType, proto.Representation()
 }
@@ -107,8 +114,84 @@ var _ schemaDefinition = (*ModuleBlock)(nil)
 //
 // A single LensBlock maybe referenced by an unlimited number of lens modules/configurations.
 type LensBlock struct {
+	// If the total number of bytes is deemed too large for ipfs to transport we chunk the block
+	// into multiple child blocks.
+	//
+	// This is governed by our `maxBlockSize` parameter/option.
+	Chunks *Chunks
+
 	// WasmBytes is the executable wasm binary of the lens.
-	WasmBytes []byte
+	//
+	// Or, if the bytes are too large for ipfs to transport, a chunk of those bytes.
+	WasmBytes *[]byte
+}
+
+var _ schemaDefinition = (*LensBlock)(nil)
+
+type Chunks []datamodel.Link
+
+var _ schemaDefinition = (*Chunks)(nil)
+
+func storeLensBlock(
+	ctx context.Context,
+	linkSys *linking.LinkSystem,
+	wasmBytes []byte,
+	maxBlockSize int,
+) (datamodel.Link, error) {
+	chunkBytes := [][]byte{}
+	for chunk := range slices.Chunk(wasmBytes, maxBlockSize) {
+		chunkBytes = append(chunkBytes, chunk)
+	}
+
+	if len(chunkBytes) == 1 {
+		block := LensBlock{
+			WasmBytes: &chunkBytes[0],
+		}
+		return linkSys.Store(linking.LinkContext{Ctx: ctx}, getLinkPrototype(), block.generateNode())
+	}
+
+	links := []datamodel.Link{}
+	for _, chunk := range chunkBytes {
+		block := &LensBlock{
+			WasmBytes: &chunk,
+		}
+
+		link, err := linkSys.Store(linking.LinkContext{Ctx: ctx}, getLinkPrototype(), block.generateNode())
+		if err != nil {
+			return nil, err
+		}
+
+		links = append(links, link)
+	}
+
+	chunkLinks := Chunks(links)
+	block := LensBlock{
+		Chunks: &chunkLinks,
+	}
+	return linkSys.Store(linking.LinkContext{Ctx: ctx}, getLinkPrototype(), block.generateNode())
+}
+
+func (b *LensBlock) Bytes(ctx context.Context, linkSys *linking.LinkSystem) ([]byte, error) {
+	switch {
+	case b.WasmBytes != nil:
+		return *b.WasmBytes, nil
+	case b.Chunks != nil:
+		var buf bytes.Buffer
+		for _, chunk := range *b.Chunks {
+			lensNode, err := linkSys.Load(linking.LinkContext{Ctx: ctx}, chunk, LensBlockSchemaPrototype)
+			if err != nil {
+				return nil, err
+			}
+			lensBlock := bindnode.Unwrap(lensNode).(*LensBlock)
+			b, err := lensBlock.Bytes(ctx, linkSys)
+			if err != nil {
+				return nil, err
+			}
+			buf.Write(b)
+		}
+		return buf.Bytes(), nil
+	}
+	return nil, nil
 }
 
 var _ schemaDefinition = (*LensBlock)(nil)
@@ -137,9 +220,16 @@ func (b *ModuleBlock) IPLDSchemaBytes() []byte {
 
 func (b *LensBlock) IPLDSchemaBytes() []byte {
 	return []byte(`
-		type lensBlock struct {
-			wasmBytes Bytes
-		}
+		type lensBlock union {
+			| chunks "chunks"
+			| Bytes "wasmBytes"
+		} representation keyed
+	`)
+}
+
+func (b *Chunks) IPLDSchemaBytes() []byte {
+	return []byte(`
+		type chunks [Link]
 	`)
 }
 
@@ -185,15 +275,20 @@ func LoadLensModel(ctx context.Context, linkSys *linking.LinkSystem, cid cid.Cid
 			}
 		}
 
-		lensNode, err := linkSys.Load(ipld.LinkContext{Ctx: ctx}, moduleBlock.Lens, LensBlockSchemaPrototype)
+		lensNode, err := linkSys.Load(linking.LinkContext{Ctx: ctx}, moduleBlock.Lens, LensBlockSchemaPrototype)
 		if err != nil {
 			return model.Lens{}, err
 		}
 		lensBlock := bindnode.Unwrap(lensNode).(*LensBlock)
 
+		wasmBytes, err := lensBlock.Bytes(ctx, linkSys)
+		if err != nil {
+			return model.Lens{}, err
+		}
+
 		var path string
-		if len(lensBlock.WasmBytes) != 0 {
-			path = "data:application/octet-stream," + string(lensBlock.WasmBytes)
+		if len(wasmBytes) != 0 {
+			path = "data:application/octet-stream," + string(wasmBytes)
 		}
 
 		result.Lenses = append(result.Lenses, model.LensModule{
@@ -209,6 +304,7 @@ func LoadLensModel(ctx context.Context, linkSys *linking.LinkSystem, cid cid.Cid
 func writeConfigBlock(
 	ctx context.Context,
 	linkSys *linking.LinkSystem,
+	maxBlockSize int,
 	cfg model.Lens,
 ) (datamodel.Link, error) {
 	moduleLinks := make([]datamodel.Link, 0, len(cfg.Lenses))
@@ -218,11 +314,8 @@ func writeConfigBlock(
 		if err != nil {
 			return nil, err
 		}
-		lensBlock := LensBlock{
-			WasmBytes: wasmBytes,
-		}
 
-		lensLink, err := linkSys.Store(linking.LinkContext{Ctx: ctx}, getLinkPrototype(), lensBlock.generateNode())
+		lensLink, err := storeLensBlock(ctx, linkSys, wasmBytes, maxBlockSize)
 		if err != nil {
 			return nil, err
 		}

host-go/store/store.go

@@ -24,6 +24,8 @@ import (
 	"github.com/sourcenetwork/lens/host-go/repository"
 )
 
+const defaultBlockMaxSize int = 1024 * 1024 * 3
+
 type Store interface {
 	// Add stores the given Lens and returns its content ID.
 	//
@@ -78,13 +80,22 @@ func New(
 	runtime module.Runtime,
 	blockstoreNamespace string,
 	blockstoreChunksize immutable.Option[int],
+	maxBlockSize immutable.Option[int],
 	indexstoreNamespace string,
 ) TxnStore {
+	var maxSize int
+	if maxBlockSize.HasValue() {
+		maxSize = maxBlockSize.Value()
+	} else {
+		maxSize = defaultBlockMaxSize
+	}
+
 	return &implicitTxnStore{
 		txnSource:           txnSource,
 		repository:          repository.NewRepository(poolSize, runtime, &repositoryTxnSource{src: txnSource}),
 		blockstoreNamespace: blockstoreNamespace,
 		blockstoreChunksize: blockstoreChunksize,
+		maxBlockSize:        maxSize,
 		indexstoreNamespace: indexstoreNamespace,
 	}
 }
@@ -97,19 +108,28 @@ func NewWithRepository(
 	repository repository.TxnRepository,
 	blockstoreNamespace string,
 	blockstoreChunksize immutable.Option[int],
+	maxBlockSize immutable.Option[int],
 	indexstoreNamespace string,
 ) TxnStore {
+	var maxSize int
+	if maxBlockSize.HasValue() {
+		maxSize = maxBlockSize.Value()
+	} else {
+		maxSize = defaultBlockMaxSize
+	}
+
 	return &implicitTxnStore{
 		txnSource:           txnSource,
 		repository:          repository,
 		blockstoreNamespace: blockstoreNamespace,
 		blockstoreChunksize: blockstoreChunksize,
+		maxBlockSize:        maxSize,
 		indexstoreNamespace: indexstoreNamespace,
 	}
 }
 
 func add(ctx context.Context, cfg model.Lens, txn *txn) (cid.Cid, error) {
-	configLink, err := writeConfigBlock(ctx, txn.linkSystem, cfg)
+	configLink, err := writeConfigBlock(ctx, txn.linkSystem, txn.maxBlockSize, cfg)
 	if err != nil {
 		return cid.Undef, err
 	}

host-go/store/txn.go

@@ -23,9 +23,10 @@ type Txn interface {
 
 type txn struct {
 	repository.Txn
-	linkSystem *linking.LinkSystem
-	indexstore corekv.ReaderWriter
-	repository repository.Repository
+	linkSystem   *linking.LinkSystem
+	maxBlockSize int
+	indexstore   corekv.ReaderWriter
+	repository   repository.Repository
 }
 
 // repositoryTxnSource wraps a `TxnSource` so that it satisfies the `repository.TxnSource`

host-go/store/txn_store.go

@@ -24,6 +24,7 @@ type implicitTxnStore struct {
 	blockstoreNamespace string
 	blockstoreChunksize immutable.Option[int]
 	indexstoreNamespace string
+	maxBlockSize        int
 }
 
 type explicitTxnStore struct {
@@ -63,10 +64,11 @@ func (s *implicitTxnStore) wrapTxn(t Txn) *txn {
 	}
 
 	return &txn{
-		Txn:        t,
-		linkSystem: makeLinkSystem(bStore),
-		indexstore: namespace.Wrap(t, []byte(s.indexstoreNamespace)),
-		repository: s.repository.WithTxn(t),
+		Txn:          t,
+		maxBlockSize: s.maxBlockSize,
+		linkSystem:   makeLinkSystem(bStore),
+		indexstore:   namespace.Wrap(t, []byte(s.indexstoreNamespace)),
+		repository:   s.repository.WithTxn(t),
 	}
 }