Add availability layer demo

Demo allows storing and retrieving blocks in the availability
layer.

samples/availability-layer-demo/client.go

@@ -0,0 +1,143 @@
+package main
+
+import (
+	"context"
+	"crypto"
+	"fmt"
+	"os"
+
+	"gopkg.in/alecthomas/kingpin.v2"
+
+	"github.com/filecoin-project/mir"
+	"github.com/filecoin-project/mir/pkg/availability/multisigcollector"
+	mirCrypto "github.com/filecoin-project/mir/pkg/crypto"
+	"github.com/filecoin-project/mir/pkg/logging"
+	"github.com/filecoin-project/mir/pkg/mempool/simplemempool"
+	"github.com/filecoin-project/mir/pkg/modules"
+	"github.com/filecoin-project/mir/pkg/net/grpc"
+	t "github.com/filecoin-project/mir/pkg/types"
+	grpctools "github.com/filecoin-project/mir/pkg/util/grpc"
+)
+
+const (
+
+	// Base port number for the nodes to listen to messages from each other.
+	// The nodes will listen on ports starting from nodeBasePort through nodeBasePort+3.
+	nodeBasePort = 10000
+
+	// The number of nodes participating in the chat.
+	nodeNumber = 4
+)
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	args := parseArgs(os.Args)
+
+	// IDs of nodes that are part of the system.
+	// This example uses a static configuration of nodeNumber nodes.
+	nodeIDs := make([]t.NodeID, nodeNumber)
+	for i := 0; i < nodeNumber; i++ {
+		nodeIDs[i] = t.NewNodeIDFromInt(i)
+	}
+
+	nodeAddrs := make(map[t.NodeID]t.NodeAddress)
+	for i := range nodeIDs {
+		nodeAddrs[t.NewNodeIDFromInt(i)] = t.NodeAddress(grpctools.NewDummyMultiaddr(i + nodeBasePort))
+	}
+
+	transport, err := grpc.NewTransport(args.OwnID, nodeAddrs[args.OwnID], logging.NilLogger)
+	if err != nil {
+		return fmt.Errorf("failed to get network transport %w", err)
+	}
+	if err := transport.Start(); err != nil {
+		return fmt.Errorf("could not start network transport: %w", err)
+	}
+	transport.Connect(context.Background(), nodeAddrs)
+
+	mempool := simplemempool.NewModule(
+		&simplemempool.ModuleConfig{
+			Self:   "mempool",
+			Hasher: "hasher",
+		},
+		&simplemempool.ModuleParams{
+			MaxTransactionsInBatch: 10,
+		},
+	)
+
+	availability, err := multisigcollector.NewModule(
+		&multisigcollector.ModuleConfig{
+			Self:    "availability",
+			Mempool: "mempool",
+			Net:     "net",
+			Crypto:  "crypto",
+		},
+		&multisigcollector.ModuleParams{
+			InstanceUID: []byte("testing instance"),
+			AllNodes:    nodeIDs,
+			F:           (len(nodeIDs) - 1) / 2,
+		},
+		args.OwnID,
+	)
+	if err != nil {
+		return err
+	}
+
+	// control module reads the user input from the console and processes it.
+	control := newControlModule()
+
+	m := map[t.ModuleID]modules.Module{
+		"net":          transport,
+		"mempool":      mempool,
+		"hasher":       mirCrypto.NewHasher(crypto.SHA256),
+		"crypto":       mirCrypto.New(&mirCrypto.DummyCrypto{DummySig: []byte{0}}),
+		"availability": availability,
+		"control":      control,
+	}
+
+	// create a Mir node
+	node, err := mir.NewNode("client", &mir.NodeConfig{Logger: logging.NilLogger}, m, nil, nil)
+	if err != nil {
+		return fmt.Errorf("error creating a Mir node: %w", err)
+	}
+
+	// run the node
+	err = node.Run(context.Background())
+	if err != nil {
+		return fmt.Errorf("error running node: %w", err)
+	}
+
+	return nil
+}
+
+// Parses the command-line arguments and returns them in a params struct.
+func parseArgs(args []string) *parsedArgs {
+	app := kingpin.New("chat-demo", "Small chat application to demonstrate the usage of the Mir library.")
+	verbose := app.Flag("verbose", "Verbose mode.").Short('v').Bool()
+	ownID := app.Arg("id", "ID of this node").Required().String()
+
+	if _, err := app.Parse(args[1:]); err != nil { // Skip args[0], which is the name of the program, not an argument.
+		app.FatalUsage("could not parse arguments: %v\n", err)
+	}
+
+	return &parsedArgs{
+		OwnID:   t.NodeID(*ownID),
+		Verbose: *verbose,
+	}
+}
+
+// parsedArgs represents parsed command-line parameters passed to the program.
+type parsedArgs struct {
+
+	// ID of this node.
+	// The package github.com/hyperledger-labs/mir/pkg/types defines this and other types used by the library.
+	OwnID t.NodeID
+
+	// If set, print verbose output to stdout.
+	Verbose bool
+}

samples/availability-layer-demo/control-module.go

@@ -0,0 +1,165 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"encoding/base64"
+	"fmt"
+	"os"
+	"strings"
+
+	"google.golang.org/protobuf/proto"
+
+	availabilityevents "github.com/filecoin-project/mir/pkg/availability/events"
+	"github.com/filecoin-project/mir/pkg/events"
+	"github.com/filecoin-project/mir/pkg/modules"
+	"github.com/filecoin-project/mir/pkg/pb/availabilitypb"
+	"github.com/filecoin-project/mir/pkg/pb/eventpb"
+	"github.com/filecoin-project/mir/pkg/pb/requestpb"
+)
+
+type controlModule struct {
+	eventsOut           chan *events.EventList
+	readyForNextCommand chan struct{}
+}
+
+func newControlModule() modules.ActiveModule {
+	return &controlModule{
+		eventsOut: make(chan *events.EventList),
+	}
+}
+
+func (m *controlModule) ImplementsModule() {}
+
+func (m *controlModule) ApplyEvents(ctx context.Context, events *events.EventList) error {
+	iter := events.Iterator()
+	for event := iter.Next(); event != nil; event = iter.Next() {
+		switch event := event.Type.(type) {
+
+		case *eventpb.Event_Init:
+			go func() {
+				err := m.readConsole()
+				if err != nil {
+					panic(err)
+				}
+			}()
+
+		case *eventpb.Event_Availability:
+			switch event := event.Availability.Type.(type) {
+
+			case *availabilitypb.Event_NewCert:
+				certBytes, err := proto.Marshal(event.NewCert.Cert)
+				if err != nil {
+					return fmt.Errorf("error marshalling certificate: %w", err)
+				}
+
+				fmt.Println(base64.StdEncoding.EncodeToString(certBytes))
+				close(m.readyForNextCommand)
+
+			case *availabilitypb.Event_ProvideTransactions:
+				for _, tx := range event.ProvideTransactions.Txs {
+					fmt.Println(string(tx.Data))
+				}
+				close(m.readyForNextCommand)
+			}
+
+		}
+	}
+
+	return nil
+}
+
+func (m *controlModule) EventsOut() <-chan *events.EventList {
+	return m.eventsOut
+}
+
+func (m *controlModule) readConsole() error {
+	// Read the user input
+	scanner := bufio.NewScanner(os.Stdin)
+
+	for {
+		fmt.Println("Type in the command ('createBatch', 'readBatch')")
+		scanner.Scan()
+		if scanner.Err() != nil {
+			return fmt.Errorf("error reading from console: %w", scanner.Err())
+		}
+
+		text := scanner.Text()
+
+		switch cmd := strings.TrimSpace(text); cmd {
+		case "createBatch":
+			m.readyForNextCommand = make(chan struct{})
+			err := m.createBatch(scanner)
+			if err != nil {
+				return err
+			}
+			<-m.readyForNextCommand
+
+		case "readBatch":
+			m.readyForNextCommand = make(chan struct{})
+			err := m.readBatch(scanner)
+			if err != nil {
+				return err
+			}
+			<-m.readyForNextCommand
+
+		default:
+			fmt.Println("Unknown command: ", cmd)
+		}
+	}
+}
+
+func (m *controlModule) createBatch(scanner *bufio.Scanner) error {
+	fmt.Println("Type in 1 transaction per line, then type 'send!' and press Enter")
+
+	for {
+		scanner.Scan()
+		if scanner.Err() != nil {
+			return fmt.Errorf("error reading user data: %w", scanner.Err())
+		}
+
+		text := scanner.Text()
+		if strings.TrimSpace(text) == "send!" {
+			break
+		}
+
+		request := &requestpb.Request{Data: []byte(text)}
+		m.eventsOut <- events.ListOf(events.NewClientRequests("mempool", []*requestpb.Request{request}))
+	}
+
+	m.eventsOut <- events.ListOf(availabilityevents.RequestCert("availability", &availabilitypb.RequestCertOrigin{
+		Module: "control",
+		Type:   &availabilitypb.RequestCertOrigin_ContextStore{},
+	}))
+
+	return nil
+}
+
+func (m *controlModule) readBatch(scanner *bufio.Scanner) error {
+	fmt.Println("type in the availability certificate and press Enter")
+
+	scanner.Scan()
+	if scanner.Err() != nil {
+		return fmt.Errorf("error reading batch id: %w", scanner.Err())
+	}
+
+	certBase64 := strings.TrimSpace(scanner.Text())
+	certBytes, err := base64.StdEncoding.DecodeString(certBase64)
+	if err != nil {
+		return fmt.Errorf("error decoding certificate: %w", err)
+	}
+
+	cert := new(availabilitypb.Cert)
+	err = proto.Unmarshal(certBytes, cert)
+	if err != nil {
+		return fmt.Errorf("error unmarshalling certificate: %w", err)
+	}
+
+	m.eventsOut <- events.ListOf(availabilityevents.RequestTransactions("availability", cert,
+		&availabilitypb.RequestTransactionsOrigin{
+			Module: "control",
+			Type:   &availabilitypb.RequestTransactionsOrigin_ContextStore{},
+		}))
+
+	return nil
+}

samples/availability-layer-demo/run.sh

@@ -0,0 +1,19 @@
+NODE_0_LOG="./node_0.log"
+NODE_1_LOG="./node_1.log"
+NODE_2_LOG="./node_2.log"
+NODE_3_LOG="./node_3.log"
+
+rm -rf ./node_*.log
+
+tmux new-session -d -s "demo" \; \
+  new-window   -t "demo" \; \
+  \
+  split-window -t "demo:0" -v \; \
+  split-window -t "demo:0.0" -h \; \
+  split-window -t "demo:0.2" -h \; \
+  \
+  send-keys -t "demo:0.0" "go run ./samples/availability-layer-demo 0 2>&1 | tee $NODE_0_LOG" Enter \; \
+  send-keys -t "demo:0.1" "go run ./samples/availability-layer-demo 1 2>&1 | tee $NODE_1_LOG" Enter \; \
+  send-keys -t "demo:0.2" "go run ./samples/availability-layer-demo 2 2>&1 | tee $NODE_2_LOG" Enter \; \
+  send-keys -t "demo:0.3" "go run ./samples/availability-layer-demo 3 2>&1 | tee $NODE_3_LOG" Enter \; \
+  attach-session -t "demo:0.0"