Refactor node (#54)

Author: mooselumph

.gitignore

@@ -1,2 +1,3 @@
+inabox/testdata/*
 test/testdata/*
 test/resources/kzg/SRSTables/*

core/attestation.go

@@ -21,6 +21,7 @@ type Signature interface {
 	Add(Signature) Signature
 	Verify(PubKeyG2, [32]byte) bool
 	Serialize() [32]byte
+	Deserialize([]byte) Signature
 }
 
 type PubKeyG1 interface {
@@ -66,6 +67,8 @@ var _ SignatureAggregator = (*StdSignatureAggregator)(nil)
 
 func (a *StdSignatureAggregator) AggregateSignatures(state *OperatorState, quorumParams []QuorumParam, message [32]byte, messageChan chan SignerMessage) (*SignatureAggregation, error) {
 
+	// TODO: Add logging
+
 	// Ensure all quorums are found in state
 	for _, quorum := range quorumParams {
 		_, found := state.QuorumInfo[quorum.QuorumID]

core/bn254/attestation.go

@@ -82,6 +82,15 @@ func (s Signature) Serialize() [32]byte {
 	return s.G1Affine.Bytes()
 }
 
+func (s Signature) Deserialize(data []byte) core.Signature {
+	p := new(bn254.G1Affine)
+	_, err := p.SetBytes(data)
+	if err != nil {
+		return nil
+	}
+	return Signature{p}
+}
+
 type PubKeyG1 struct {
 	*bn254.G1Affine
 }

core/mock/state.go

@@ -17,8 +17,10 @@ type ChainDataMock struct {
 var _ core.ChainState = (*ChainDataMock)(nil)
 
 type PrivateOperatorInfo struct {
-	OperatorInfo *core.OperatorInfo
-	KeyPair      core.KeyPair
+	*core.OperatorInfo
+	KeyPair core.KeyPair
+	Host    string
+	Port    string
 }
 
 type PrivateOperatorState struct {
@@ -65,8 +67,11 @@ func (d *ChainDataMock) GetTotalOperatorState(blockNumber uint) (*core.OperatorS
 
 		stake := ind + 1
 
+		host := "0.0.0.0"
+		port := fmt.Sprintf("3%03v", int(ind))
+
 		op := &core.OperatorInfo{
-			Socket:          fmt.Sprintf("0.0.0.0:%v", ind),
+			Socket:          fmt.Sprintf("%v:%v", host, port),
 			PubkeyG1:        d.KeyPairs[ind].GetPubKeyG1(),
 			PubkeyG2:        d.KeyPairs[ind].GetPubKeyG2(),
 			FromBlockNumber: 0,
@@ -76,6 +81,8 @@ func (d *ChainDataMock) GetTotalOperatorState(blockNumber uint) (*core.OperatorS
 		private := &PrivateOperatorInfo{
 			OperatorInfo: op,
 			KeyPair:      d.KeyPairs[ind],
+			Host:         host,
+			Port:         port,
 		}
 
 		id := core.OperatorId(common.HexToAddress(big.NewInt(int64(ind + 1)).Text(16)))

core/validator.go

@@ -17,8 +17,13 @@ type ChunkValidator struct {
 	Id         OperatorId
 }
 
-func NewChunkValidator() *ChunkValidator {
-	return &ChunkValidator{}
+func NewChunkValidator(enc Encoder, asgn AssignmentCoordinator, cst ChainState, id OperatorId) *ChunkValidator {
+	return &ChunkValidator{
+		Encoder:    enc,
+		Assignment: asgn,
+		ChainState: cst,
+		Id:         id,
+	}
 }
 
 func (v *ChunkValidator) ValidateChunks(chunks []Chunk, glom GlomHeader, batch BatchHeader) error {

disperser/batcher.go

@@ -3,7 +3,6 @@ package disperser
 import (
 	"context"
 	"errors"
-	"fmt"
 	"time"
 
 	"github.com/Layr-Labs/eigenda/common"
@@ -76,8 +75,6 @@ func NewBatcher(
 
 func (b *Batcher) Start(ctx context.Context) {
 
-	fmt.Println("Starting batcher...")
-
 	go func() {
 
 		log := b.Logger
@@ -292,6 +289,7 @@ func (b *Batcher) ConstructGlom(blobs []*core.Blob, plan *GlomPlan, quorum Quoru
 		Commitment:         commits.Commitment,
 		LengthProof:        commits.LengthProof,
 		Length:             commits.Length,
+		ChunkLength:        plan.ChunkLength,
 		QuantizationFactor: quorum.QuantizationFactor,
 		NumBlobs:           1,
 		SecurityParam:      plan.SecurityParam,
@@ -311,10 +309,10 @@ func (b *Batcher) ConstructBatch(state *core.OperatorState, blobs []*core.Blob,
 	batchBundles := make(map[core.OperatorId][][]core.Chunk, plan.NumOperators)
 	gloms := make([]core.GlomHeader, len(plan.Gloms))
 
-	for _, glom := range plan.Gloms {
+	for ind, glomPlan := range plan.Gloms {
 
 		// Construct the glom
-		glomBundles, glom, err := b.ConstructGlom(blobs, &glom, plan.Quorums[glom.QuorumID])
+		glomBundles, glom, err := b.ConstructGlom(blobs, &glomPlan, plan.Quorums[glomPlan.QuorumID])
 		if err != nil {
 			return nil, nil, err
 		}
@@ -328,7 +326,7 @@ func (b *Batcher) ConstructBatch(state *core.OperatorState, blobs []*core.Blob,
 			}
 
 		}
-		gloms = append(gloms, *glom)
+		gloms[ind] = *glom
 	}
 
 	chunkBatches := make(map[core.OperatorId]core.ChunkBatch, len(batchBundles))

disperser/cmd/basic/config.go

@@ -34,7 +34,7 @@ func NewConfig(ctx *cli.Context) Config {
 			PullInterval: ctx.GlobalDuration(flags.PullIntervalFlag.Name),
 		},
 		BLSOperatorStateRetrieverAddr: ctx.GlobalString(flags.BlsOperatorStateRetrieverFlag.Name),
-		EigenDAServiceManagerAddr:     ctx.GlobalString(flags.EigenDAServiceManageFlag.Name),
+		EigenDAServiceManagerAddr:     ctx.GlobalString(flags.EigenDAServiceManagerFlag.Name),
 		Address:                       ctx.GlobalString(flags.AddressFlag.Name),
 	}
 	return config

disperser/cmd/basic/flags/flags.go

@@ -24,17 +24,19 @@ var (
 		Required: true,
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "GRPC_PORT"),
 	}
-	BlsOperatorStateRetrieverFlag = cli.StringFlag{
-		Name:  "bls-operator-state-retriever",
-		Usage: "Address of the BLS Operator State Retriever",
-	}
 	AddressFlag = cli.StringFlag{
 		Name:     "address",
 		Usage:    "Address of the disperser",
 		Required: true,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "Address"),
+	}
+	BlsOperatorStateRetrieverFlag = cli.StringFlag{
+		Name:     "bls-operator-state-retriever",
+		Usage:    "Address of the BLS Operator State Retriever",
+		Required: true,
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "BLS_OPERATOR_STATE_RETRIVER"),
 	}
-	EigenDAServiceManageFlag = cli.StringFlag{
+	EigenDAServiceManagerFlag = cli.StringFlag{
 		Name:     "eigenda-service-manager",
 		Usage:    "Address of the EigenDA Service Manager",
 		Required: true,
@@ -46,7 +48,7 @@ var requiredFlags = []cli.Flag{
 	PullIntervalFlag,
 	GrpcPortFlag,
 	BlsOperatorStateRetrieverFlag,
-	EigenDAServiceManageFlag,
+	EigenDAServiceManagerFlag,
 	AddressFlag,
 }
 

disperser/grpc/dispatcher.go

@@ -7,6 +7,7 @@ import (
 	"github.com/Layr-Labs/eigenda/api/grpc/node"
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/Layr-Labs/eigenda/core"
+	"github.com/Layr-Labs/eigenda/core/bn254"
 	"github.com/Layr-Labs/eigenda/disperser"
 
 	"google.golang.org/grpc"
@@ -60,12 +61,12 @@ func (c *Dispatcher) sendAllChunks(ctx context.Context, state *core.OperatorStat
 					Signature: nil,
 					Operator:  id,
 				}
-			}
-
-			update <- core.SignerMessage{
-				Signature: sig,
-				Operator:  id,
-				Err:       nil,
+			} else {
+				update <- core.SignerMessage{
+					Signature: sig,
+					Operator:  id,
+					Err:       nil,
+				}
 			}
 
 		}(op, id)
@@ -76,6 +77,7 @@ func (c *Dispatcher) sendAllChunks(ctx context.Context, state *core.OperatorStat
 func (c *Dispatcher) sendChunks(ctx context.Context, chunkBatch core.ChunkBatch, header *core.BatchHeader, op *core.OperatorInfo) (core.Signature, error) {
 
 	// TODO Add secure Grpc
+
 	conn, err := grpc.Dial(
 		op.Socket,
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
@@ -98,13 +100,14 @@ func (c *Dispatcher) sendChunks(ctx context.Context, chunkBatch core.ChunkBatch,
 	opt := grpc.MaxCallSendMsgSize(1024 * 1024 * 300)
 	reply, err := gc.StoreChunks(ctx, request, opt)
 
-	if err == nil {
-		sig := reply.GetSignature()
-		_ = sig
-		return nil, nil
-	} else {
+	if err != nil {
 		return nil, err
 	}
+
+	sigBytes := reply.GetSignature()
+	sig := new(bn254.Signature).Deserialize(sigBytes)
+	return sig, nil
+
 }
 
 func getStoreChunksRequest(chunkBatch core.ChunkBatch, header *core.BatchHeader) (*node.StoreChunksRequest, error) {