Merge pull request #89 from ethersphere/network-testing-framework-snapshot

p2p/simulations: Snapshot support

p2p/simulations/adapters/docker.go

@@ -47,18 +47,19 @@ func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
 	}
 
 	// generate the config
-	conf := node.DefaultConfig
+	conf := &execNodeConfig{
-	conf.DataDir = "/data"
+		Stack: node.DefaultConfig,
-	conf.P2P.EnableMsgEvents = true
+		Node:  config,
-	conf.P2P.NoDiscovery = true
+	}
-	conf.P2P.NAT = nil
+	conf.Stack.DataDir = "/data"
+	conf.Stack.P2P.EnableMsgEvents = true
+	conf.Stack.P2P.NoDiscovery = true
+	conf.Stack.P2P.NAT = nil
 
 	node := &DockerNode{
 		ExecNode: ExecNode{
-			ID:      config.Id,
+			ID:     config.Id,
-			Service: config.Service,
+			Config: conf,
-			Config:  &conf,
-			key:     config.PrivateKey,
 		},
 	}
 	node.newCmd = node.dockerCommand
@@ -80,8 +81,8 @@ func (n *DockerNode) dockerCommand() *exec.Cmd {
 	return exec.Command(
 		"sh", "-c",
 		fmt.Sprintf(
-			`exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" --env _P2P_NODE_KEY="${_P2P_NODE_KEY}" %s p2p-node %s %s`,
+			`exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" %s p2p-node %s %s`,
-			dockerImage, n.Service, n.ID.String(),
+			dockerImage, n.Config.Node.Service, n.ID.String(),
 		),
 	)
 }

p2p/simulations/adapters/exec.go

@@ -3,7 +3,6 @@ package adapters
 import (
 	"context"
 	"crypto/ecdsa"
-	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -18,7 +17,6 @@ import (
 	"time"
 
 	"github.com/docker/docker/pkg/reexec"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
@@ -60,22 +58,23 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
 	}
 
 	// generate the config
-	conf := node.DefaultConfig
+	conf := &execNodeConfig{
-	conf.DataDir = filepath.Join(dir, "data")
+		Stack: node.DefaultConfig,
-	conf.P2P.EnableMsgEvents = true
+		Node:  config,
-	conf.P2P.NoDiscovery = true
+	}
-	conf.P2P.NAT = nil
+	conf.Stack.DataDir = filepath.Join(dir, "data")
+	conf.Stack.P2P.EnableMsgEvents = true
+	conf.Stack.P2P.NoDiscovery = true
+	conf.Stack.P2P.NAT = nil
 
 	// listen on a random localhost port (we'll get the actual port after
 	// starting the node through the RPC admin.nodeInfo method)
-	conf.P2P.ListenAddr = "127.0.0.1:0"
+	conf.Stack.P2P.ListenAddr = "127.0.0.1:0"
 
 	node := &ExecNode{
-		ID:      config.Id,
+		ID:     config.Id,
-		Service: config.Service,
+		Dir:    dir,
-		Dir:     dir,
+		Config: conf,
-		Config:  &conf,
-		key:     config.PrivateKey,
 	}
 	node.newCmd = node.execCommand
 	return node, nil
@@ -89,12 +88,11 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
 // (so for example we can run the node in a remote Docker container and
 // still communicate with it).
 type ExecNode struct {
-	ID      *NodeId
+	ID     *NodeId
-	Service string
+	Dir    string
-	Dir     string
+	Config *execNodeConfig
-	Config  *node.Config
+	Cmd    *exec.Cmd
-	Cmd     *exec.Cmd
+	Info   *p2p.NodeInfo
-	Info    *p2p.NodeInfo
 
 	client *rpc.Client
 	rpcMux *rpcMux
@@ -116,11 +114,10 @@ func (n *ExecNode) Client() (*rpc.Client, error) {
 	return n.client, nil
 }
 
-// Start exec's the node passing the ID and service as command line arguments,
+// Start exec's the node passing the ID and service as command line arguments
-// the node config encoded as JSON in the _P2P_NODE_CONFIG environment
+// and the node config encoded as JSON in the _P2P_NODE_CONFIG environment
-// variable and the node's private key hex-endoded in the _P2P_NODE_KEY
+// variable
-// environment variable
+func (n *ExecNode) Start(snapshot []byte) (err error) {
-func (n *ExecNode) Start() (err error) {
 	if n.Cmd != nil {
 		return errors.New("already started")
 	}
@@ -131,15 +128,14 @@ func (n *ExecNode) Start() (err error) {
 		}
 	}()
 
-	// encode the config
+	// encode a copy of the config containing the snapshot
-	conf, err := json.Marshal(n.Config)
+	confCopy := *n.Config
+	confCopy.Snapshot = snapshot
+	confData, err := json.Marshal(confCopy)
 	if err != nil {
 		return fmt.Errorf("error generating node config: %s", err)
 	}
 
-	// encode the private key
-	key := hex.EncodeToString(crypto.FromECDSA(n.key))
-
 	// create a net.Pipe for RPC communication over stdin / stdout
 	pipe1, pipe2 := net.Pipe()
 
@@ -148,10 +144,7 @@ func (n *ExecNode) Start() (err error) {
 	cmd.Stdin = pipe1
 	cmd.Stdout = pipe1
 	cmd.Stderr = os.Stderr
-	cmd.Env = append(os.Environ(),
+	cmd.Env = append(os.Environ(), fmt.Sprintf("_P2P_NODE_CONFIG=%s", confData))
-		fmt.Sprintf("_P2P_NODE_CONFIG=%s", conf),
-		fmt.Sprintf("_P2P_NODE_KEY=%s", key),
-	)
 	if err := cmd.Start(); err != nil {
 		return fmt.Errorf("error starting node: %s", err)
 	}
@@ -177,7 +170,7 @@ func (n *ExecNode) Start() (err error) {
 func (n *ExecNode) execCommand() *exec.Cmd {
 	return &exec.Cmd{
 		Path: reexec.Self(),
-		Args: []string{"p2p-node", n.Service, n.ID.String()},
+		Args: []string{"p2p-node", n.Config.Node.Service, n.ID.String()},
 	}
 }
 
@@ -214,12 +207,11 @@ func (n *ExecNode) Stop() error {
 
 // NodeInfo returns information about the node
 func (n *ExecNode) NodeInfo() *p2p.NodeInfo {
-	if n.client == nil {
+	info := &p2p.NodeInfo{
-		return n.Info
+		ID: n.ID.String(),
 	}
-	info := &p2p.NodeInfo{}
+	if n.client != nil {
-	if err := n.client.Call(&info, "admin_nodeInfo"); err != nil {
+		n.client.Call(&info, "admin_nodeInfo")
-		return n.Info
 	}
 	return info
 }
@@ -234,16 +226,33 @@ func (n *ExecNode) ServeRPC(conn net.Conn) error {
 	return nil
 }
 
+// Snapshot creates a snapshot of the service state by calling the
+// simulation_snapshot RPC method
+func (n *ExecNode) Snapshot() ([]byte, error) {
+	if n.client == nil {
+		return nil, errors.New("RPC not started")
+	}
+	var snapshot []byte
+	return snapshot, n.client.Call(&snapshot, "simulation_snapshot")
+}
+
 func init() {
 	// register a reexec function to start a devp2p node when the current
 	// binary is executed as "p2p-node"
 	reexec.Register("p2p-node", execP2PNode)
 }
 
+// execNodeConfig is used to serialize the node configuration so it can be
+// passed to the child process as a JSON encoded environment variable
+type execNodeConfig struct {
+	Stack    node.Config `json:"stack"`
+	Node     *NodeConfig `json:"node"`
+	Snapshot []byte      `json:"snapshot,omitempty"`
+}
+
 // execP2PNode starts a devp2p node when the current binary is executed with
-// argv[0] being "p2p-node", reading the service / ID from argv[1] / argv[2],
+// argv[0] being "p2p-node", reading the service / ID from argv[1] / argv[2]
-// the node config from the _P2P_NODE_CONFIG environment variable and the
+// and the node config from the _P2P_NODE_CONFIG environment variable
-// private key from the _P2P_NODE_KEY environment variable
 func execP2PNode() {
 	glogger := log.NewGlogHandler(log.StreamHandler(os.Stderr, log.LogfmtFormat()))
 	glogger.Verbosity(log.LvlInfo)
@@ -258,45 +267,35 @@ func execP2PNode() {
 	if confEnv == "" {
 		log.Crit("missing _P2P_NODE_CONFIG")
 	}
-	var conf node.Config
+	var conf execNodeConfig
 	if err := json.Unmarshal([]byte(confEnv), &conf); err != nil {
 		log.Crit("error decoding _P2P_NODE_CONFIG", "err", err)
 	}
-
+	conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey
-	// decode the private key
-	keyEnv := os.Getenv("_P2P_NODE_KEY")
-	if keyEnv == "" {
-		log.Crit("missing _P2P_NODE_KEY")
-	}
-	key, err := hex.DecodeString(keyEnv)
-	if err != nil {
-		log.Crit("error decoding _P2P_NODE_KEY", "err", err)
-	}
-	conf.P2P.PrivateKey = crypto.ToECDSA(key)
 
 	// initialize the service
 	serviceFunc, exists := serviceFuncs[serviceName]
 	if !exists {
 		log.Crit(fmt.Sprintf("unknown node service %q", serviceName))
 	}
-	service := serviceFunc(id)
+	service := serviceFunc(id, conf.Snapshot)
 
 	// use explicit IP address in ListenAddr so that Enode URL is usable
-	if strings.HasPrefix(conf.P2P.ListenAddr, ":") {
+	if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") {
 		addrs, err := net.InterfaceAddrs()
 		if err != nil {
 			log.Crit("error getting IP address", "err", err)
 		}
 		for _, addr := range addrs {
 			if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() {
-				conf.P2P.ListenAddr = ip.IP.String() + conf.P2P.ListenAddr
+				conf.Stack.P2P.ListenAddr = ip.IP.String() + conf.Stack.P2P.ListenAddr
 				break
 			}
 		}
 	}
 
 	// start the devp2p stack
-	stack, err := startP2PNode(&conf, service)
+	stack, err := startP2PNode(&conf.Stack, service)
 	if err != nil {
 		log.Crit("error starting p2p node", "err", err)
 	}
@@ -328,7 +327,7 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) {
 		return nil, err
 	}
 	constructor := func(ctx *node.ServiceContext) (node.Service, error) {
-		return service, nil
+		return &snapshotService{service}, nil
 	}
 	if err := stack.Register(constructor); err != nil {
 		return nil, err
@@ -339,6 +338,34 @@ func startP2PNode(conf *node.Config, service node.Service) (*node.Node, error) {
 	return stack, nil
 }
 
+// snapshotService wraps a node.Service and injects a snapshot API into the
+// list of RPC APIs
+type snapshotService struct {
+	node.Service
+}
+
+func (s *snapshotService) APIs() []rpc.API {
+	return append([]rpc.API{{
+		Namespace: "simulation",
+		Version:   "1.0",
+		Service:   SnapshotAPI{s.Service},
+	}}, s.Service.APIs()...)
+}
+
+// SnapshotAPI provides an RPC method to create a snapshot of a node.Service
+type SnapshotAPI struct {
+	service node.Service
+}
+
+func (api SnapshotAPI) Snapshot() ([]byte, error) {
+	if s, ok := api.service.(interface {
+		Snapshot() ([]byte, error)
+	}); ok {
+		return s.Snapshot()
+	}
+	return nil, nil
+}
+
 // stdioConn wraps os.Stdin / os.Stdout with a no-op Close method so we can
 // use stdio for RPC messages
 type stdioConn struct {

p2p/simulations/adapters/inproc.go

@@ -70,21 +70,13 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 	if !exists {
 		return nil, fmt.Errorf("unknown node service %q", config.Service)
 	}
-	service := serviceFunc(id)
-
-	// for simplicity, only support single protocol services (simulating
-	// multiple protocols on the same peer is extra effort, and we don't
-	// currently run any simulations which run multiple protocols)
-	if len(service.Protocols()) != 1 {
-		return nil, errors.New("service must have a single protocol")
-	}
 
 	node := &SimNode{
-		Id:        id,
+		Id:          id,
-		adapter:   s,
+		adapter:     s,
-		service:   service,
+		serviceFunc: serviceFunc,
-		peers:     make(map[discover.NodeID]MsgReadWriteCloser),
+		peers:       make(map[discover.NodeID]MsgReadWriteCloser),
-		dropPeers: make(chan struct{}),
+		dropPeers:   make(chan struct{}),
 	}
 	s.nodes[id.NodeID] = node
 	return node, nil
@@ -113,14 +105,15 @@ type MsgReadWriteCloser interface {
 // It implements the p2p.Server interface so it can be used transparently
 // by the underlying service.
 type SimNode struct {
-	lock     sync.RWMutex
+	lock        sync.RWMutex
-	Id       *NodeId
+	Id          *NodeId
-	adapter  *SimAdapter
+	adapter     *SimAdapter
-	service  node.Service
+	running     node.Service
-	peers    map[discover.NodeID]MsgReadWriteCloser
+	serviceFunc ServiceFunc
-	peerFeed event.Feed
+	peers       map[discover.NodeID]MsgReadWriteCloser
-	client   *rpc.Client
+	peerFeed    event.Feed
-	rpcMux   *rpcMux
+	client      *rpc.Client
+	rpcMux      *rpcMux
 
 	// dropPeers is used to force peer disconnects when
 	// the node is stopped
@@ -161,13 +154,35 @@ func (self *SimNode) ServeRPC(conn net.Conn) error {
 	return nil
 }
 
-// Start starts the RPC handler and the underlying service
+// Start initializes the service, starts the RPC handler and then starts
-func (self *SimNode) Start() error {
+// the service
+func (self *SimNode) Start(snapshot []byte) error {
+	service := self.serviceFunc(self.Id, snapshot)
+
+	// for simplicity, only support single protocol services (simulating
+	// multiple protocols on the same peer is extra effort, and we don't
+	// currently run any simulations which run multiple protocols)
+	if len(service.Protocols()) != 1 {
+		return errors.New("service must have a single protocol")
+	}
+
 	self.dropPeers = make(chan struct{})
-	if err := self.startRPC(); err != nil {
+	if err := self.startRPC(service); err != nil {
 		return err
 	}
-	return self.service.Start(self)
+	self.running = service
+	return service.Start(&simServer{self})
+}
+
+// simServer wraps a SimNode but modifies the Start method signature so that
+// it implements the p2p.Server interface (the Start method is never actually
+// called when using the SimAdapter)
+type simServer struct {
+	*SimNode
+}
+
+func (s *simServer) Start() error {
+	return nil
 }
 
 // Stop stops the RPC handler, stops the underlying service and disconnects
@@ -175,25 +190,24 @@ func (self *SimNode) Start() error {
 func (self *SimNode) Stop() error {
 	self.stopRPC()
 	close(self.dropPeers)
-	return self.service.Stop()
+	return self.running.Stop()
 }
 
-// Running returns whether or not the service is running by checking if the
+// Running returns whether or not the service is running
-// RPC client is set
 func (self *SimNode) Running() bool {
 	self.lock.Lock()
 	defer self.lock.Unlock()
-	return self.client != nil
+	return self.running != nil
 }
 
-// Service returns the underlying node.Service
+// Service returns the running node.Service
 func (self *SimNode) Service() node.Service {
-	return self.service
+	return self.running
 }
 
 // startRPC starts an RPC server and connects to it using an in-process RPC
 // client
-func (self *SimNode) startRPC() error {
+func (self *SimNode) startRPC(service node.Service) error {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	if self.client != nil {
@@ -202,7 +216,7 @@ func (self *SimNode) startRPC() error {
 
 	// add SimAdminAPI so that the network can call the
 	// AddPeer, RemovePeer and PeerEvents RPC methods
-	apis := append(self.service.APIs(), []rpc.API{
+	apis := append(service.APIs(), []rpc.API{
 		{
 			Namespace: "admin",
 			Version:   "1.0",
@@ -292,17 +306,21 @@ func (self *SimNode) PeerCount() int {
 
 // NodeInfo returns information about the node
 func (self *SimNode) NodeInfo() *p2p.NodeInfo {
+	self.lock.Lock()
+	defer self.lock.Unlock()
 	info := &p2p.NodeInfo{
 		ID:        self.Id.String(),
 		Enode:     self.Node().String(),
 		Protocols: make(map[string]interface{}),
 	}
-	for _, proto := range self.service.Protocols() {
+	if self.running != nil {
-		nodeInfo := interface{}("unknown")
+		for _, proto := range self.running.Protocols() {
-		if query := proto.NodeInfo; query != nil {
+			nodeInfo := interface{}("unknown")
-			nodeInfo = proto.NodeInfo()
+			if query := proto.NodeInfo; query != nil {
+				nodeInfo = proto.NodeInfo()
+			}
+			info.Protocols[proto.Name] = nodeInfo
 		}
-		info.Protocols[proto.Name] = nodeInfo
 	}
 	return info
 }
@@ -312,6 +330,17 @@ func (self *SimNode) PeersInfo() (info []*p2p.PeerInfo) {
 	return nil
 }
 
+// Snapshot creates a snapshot of the running service
+func (self *SimNode) Snapshot() ([]byte, error) {
+	self.lock.Lock()
+	service := self.running
+	self.lock.Unlock()
+	if service == nil {
+		return nil, errors.New("service not running")
+	}
+	return SnapshotAPI{service}.Snapshot()
+}
+
 // RunProtocol runs the underlying service's protocol with the peer using the
 // given MsgReadWriteCloser, emitting peer add / drop events for peer event
 // subscribers
@@ -325,7 +354,7 @@ func (self *SimNode) RunProtocol(peer *SimNode, rw MsgReadWriteCloser) {
 
 	id := peer.Id
 	log.Trace(fmt.Sprintf("protocol starting on peer %v (connection with %v)", self.Id, id))
-	protocol := self.service.Protocols()[0]
+	protocol := self.running.Protocols()[0]
 	p := p2p.NewPeer(id.NodeID, id.Label(), []p2p.Cap{})
 	go func() {
 		// emit peer add event

p2p/simulations/adapters/types.go

@@ -51,14 +51,17 @@ type Node interface {
 	// node's RPC multiplexer
 	ServeRPC(net.Conn) error
 
-	// Start starts the node
+	// Start starts the node with the given snapshot
-	Start() error
+	Start(snapshot []byte) error
 
 	// Stop stops the node
 	Stop() error
 
 	// NodeInfo returns information about the node
 	NodeInfo() *p2p.NodeInfo
+
+	// Snapshot creates a snapshot of the running service
+	Snapshot() ([]byte, error)
 }
 
 // NodeAdapter is an object which creates Nodes to be used in a simulation
@@ -202,7 +205,7 @@ func RandomNodeConfig() *NodeConfig {
 type Services map[string]ServiceFunc
 
 // ServiceFunc returns a node.Service which can be used to boot devp2p nodes
-type ServiceFunc func(id *NodeId) node.Service
+type ServiceFunc func(id *NodeId, snapshot []byte) node.Service
 
 // serviceFuncs is a map of registered services which are used to boot devp2p
 // nodes

p2p/simulations/cmd/p2psim/main.go

@@ -91,6 +91,18 @@ func main() {
 					Usage:     "stream network events",
 					Action:    streamNetwork,
 				},
+				{
+					Name:      "snapshot",
+					ArgsUsage: "<network>",
+					Usage:     "create a network snapshot to stdout",
+					Action:    createSnapshot,
+				},
+				{
+					Name:      "load",
+					ArgsUsage: "<network>",
+					Usage:     "load a network snapshot from stdin",
+					Action:    loadSnapshot,
+				},
 			},
 		},
 		{
@@ -241,6 +253,32 @@ func streamNetwork(ctx *cli.Context) error {
 	}
 }
 
+func createSnapshot(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) != 1 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	networkID := args[0]
+	snap, err := client.CreateSnapshot(networkID)
+	if err != nil {
+		return err
+	}
+	return json.NewEncoder(os.Stdout).Encode(snap)
+}
+
+func loadSnapshot(ctx *cli.Context) error {
+	args := ctx.Args()
+	if len(args) != 1 {
+		return cli.ShowCommandHelp(ctx, ctx.Command.Name)
+	}
+	networkID := args[0]
+	snap := &simulations.Snapshot{}
+	if err := json.NewDecoder(os.Stdin).Decode(snap); err != nil {
+		return err
+	}
+	return client.LoadSnapshot(networkID, snap)
+}
+
 func listNodes(ctx *cli.Context) error {
 	args := ctx.Args()
 	if len(args) != 1 {

p2p/simulations/events.go

@@ -5,48 +5,79 @@ import (
 	"time"
 )
 
+// EventType is the type of event emitted by a simulation network
 type EventType string
 
 const (
+	// EventTypeNode is the type of event emitted when a node is either
+	// created, started or stopped
 	EventTypeNode EventType = "node"
+
+	// EventTypeConn is the type of event emitted when a connection is
+	// is either established or dropped between two nodes
 	EventTypeConn EventType = "conn"
-	EventTypeMsg  EventType = "msg"
+
+	// EventTypeMsg is the type of event emitted when a p2p message it
+	// sent between two nodes
+	EventTypeMsg EventType = "msg"
 )
 
+// Event is an event emitted by a simulation network
 type Event struct {
-	Type    EventType `json:"type"`
+	// Type is the type of the event
-	Time    time.Time `json:"time"`
+	Type EventType `json:"type"`
-	Control bool      `json:"control"`
 
+	// Time is the time the event happened
+	Time time.Time `json:"time"`
+
+	// Control indicates whether the event is the result of a controlled
+	// action in the network
+	Control bool `json:"control"`
+
+	// Node is set if the type is EventTypeNode
 	Node *Node `json:"node,omitempty"`
+
+	// Conn is set if the type is EventTypeConn
 	Conn *Conn `json:"conn,omitempty"`
-	Msg  *Msg  `json:"msg,omitempty"`
+
+	// Msg is set if the type is EventTypeMsg
+	Msg *Msg `json:"msg,omitempty"`
 }
 
+// NewEvent creates a new event for the given object which should be either a
+// Node, Conn or Msg.
+//
+// The object is copied so that the event represents the state of the object
+// when NewEvent is called.
 func NewEvent(v interface{}) *Event {
 	event := &Event{Time: time.Now()}
 	switch v := v.(type) {
 	case *Node:
 		event.Type = EventTypeNode
-		event.Node = v
+		node := *v
+		event.Node = &node
 	case *Conn:
 		event.Type = EventTypeConn
-		event.Conn = v
+		conn := *v
+		event.Conn = &conn
 	case *Msg:
 		event.Type = EventTypeMsg
-		event.Msg = v
+		msg := *v
+		event.Msg = &msg
 	default:
 		panic(fmt.Sprintf("invalid event type: %T", v))
 	}
 	return event
 }
 
+// ControlEvent creates a new control event
 func ControlEvent(v interface{}) *Event {
 	event := NewEvent(v)
 	event.Control = true
 	return event
 }
 
+// String returns the string representation of the event
 func (e *Event) String() string {
 	switch e.Type {
 	case EventTypeNode:

p2p/simulations/examples/connectivity.go

@@ -26,7 +26,7 @@ func main() {
 	log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
 
 	services := map[string]adapters.ServiceFunc{
-		"ping-pong": func(id *adapters.NodeId) node.Service {
+		"ping-pong": func(id *adapters.NodeId, snapshot []byte) node.Service {
 			return newPingPongService(id)
 		},
 	}
@@ -42,7 +42,7 @@ func main() {
 
 	case "sim":
 		log.Info("using sim adapter")
-		config.Adapter = adapters.NewSimAdapter(services)
+		config.NewAdapter = func() adapters.NodeAdapter { return adapters.NewSimAdapter(services) }
 
 	case "exec":
 		tmpdir, err := ioutil.TempDir("", "p2p-example")
@@ -51,15 +51,15 @@ func main() {
 		}
 		defer os.RemoveAll(tmpdir)
 		log.Info("using exec adapter", "tmpdir", tmpdir)
-		config.Adapter = adapters.NewExecAdapter(tmpdir)
+		config.NewAdapter = func() adapters.NodeAdapter { return adapters.NewExecAdapter(tmpdir) }
 
 	case "docker":
 		log.Info("using docker adapter")
-		var err error
+		adapter, err := adapters.NewDockerAdapter()
-		config.Adapter, err = adapters.NewDockerAdapter()
 		if err != nil {
 			log.Crit("error creating docker adapter", "err", err)
 		}
+		config.NewAdapter = func() adapters.NodeAdapter { return adapter }
 
 	default:
 		log.Crit(fmt.Sprintf("unknown node adapter %q", *adapter))

p2p/simulations/http.go

@@ -59,6 +59,17 @@ func (c *Client) GetNetwork(networkID string) (*Network, error) {
 	return network, c.Get(fmt.Sprintf("/networks/%s", networkID), network)
 }
 
+// CreateSnapshot creates a network snapshot
+func (c *Client) CreateSnapshot(networkID string) (*Snapshot, error) {
+	snap := &Snapshot{}
+	return snap, c.Post(fmt.Sprintf("/networks/%s/snapshots/create", networkID), nil, snap)
+}
+
+// LoadSnapshot loads a snapshot into a network
+func (c *Client) LoadSnapshot(networkID string, snap *Snapshot) error {
+	return c.Post(fmt.Sprintf("/networks/%s/snapshots/load", networkID), snap, nil)
+}
+
 // SubscribeNetwork subscribes to network events which are sent from the server
 // as a server-sent-events stream
 func (c *Client) SubscribeNetwork(networkID string, events chan *Event) (event.Subscription, error) {
@@ -225,8 +236,9 @@ func (c *Client) Send(method, path string, in, out interface{}) error {
 
 // ServerConfig is the configuration used to start an API server
 type ServerConfig struct {
-	// Adapter is the NodeAdapter to use when creating new networks
+	// NewAdapter is called to create a new NodeAdapter for each new
-	Adapter adapters.NodeAdapter
+	// network
+	NewAdapter func() adapters.NodeAdapter
 
 	// Mocker is the function which will be called when a client sends a
 	// POST request to /networks/<netid>/mock and is expected to
@@ -246,8 +258,8 @@ type Server struct {
 
 // NewServer returns a new simulation API server
 func NewServer(config *ServerConfig) *Server {
-	if config.Adapter == nil {
+	if config.NewAdapter == nil {
-		panic("Adapter not set")
+		panic("NewAdapter not set")
 	}
 
 	s := &Server{
@@ -260,6 +272,8 @@ func NewServer(config *ServerConfig) *Server {
 	s.GET("/networks", s.GetNetworks)
 	s.GET("/networks/:netid", s.GetNetwork)
 	s.GET("/networks/:netid/events", s.StreamNetworkEvents)
+	s.POST("/networks/:netid/snapshots/create", s.CreateSnapshot)
+	s.POST("/networks/:netid/snapshots/load", s.LoadSnapshot)
 	s.POST("/networks/:netid/mock", s.StartMocker)
 	s.POST("/networks/:netid/nodes", s.CreateNode)
 	s.GET("/networks/:netid/nodes", s.GetNodes)
@@ -290,7 +304,7 @@ func (s *Server) CreateNetwork(w http.ResponseWriter, req *http.Request) {
 		if _, exists := s.networks[config.Id]; exists {
 			return nil, fmt.Errorf("network exists: %s", config.Id)
 		}
-		network := NewNetwork(s.Adapter, config)
+		network := NewNetwork(s.NewAdapter(), config)
 		s.networks[config.Id] = network
 		return network, nil
 	}()
@@ -381,6 +395,37 @@ func (s *Server) StreamNetworkEvents(w http.ResponseWriter, req *http.Request) {
 	}
 }
 
+// CreateSnapshot creates a network snapshot
+func (s *Server) CreateSnapshot(w http.ResponseWriter, req *http.Request) {
+	network := req.Context().Value("network").(*Network)
+
+	snap, err := network.Snapshot()
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	s.JSON(w, http.StatusOK, snap)
+}
+
+// LoadSnapshot loads a snapshot into a network
+func (s *Server) LoadSnapshot(w http.ResponseWriter, req *http.Request) {
+	network := req.Context().Value("network").(*Network)
+
+	snap := &Snapshot{}
+	if err := json.NewDecoder(req.Body).Decode(snap); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	if err := network.Load(snap); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	s.JSON(w, http.StatusOK, network)
+}
+
 // CreateNode creates a node in a network using the given configuration
 func (s *Server) CreateNode(w http.ResponseWriter, req *http.Request) {
 	network := req.Context().Value("network").(*Network)

p2p/simulations/http_test.go

@@ -2,6 +2,8 @@ package simulations
 
 import (
 	"context"
+	"fmt"
+	"math/rand"
 	"net/http/httptest"
 	"sync/atomic"
 	"testing"
@@ -16,10 +18,15 @@ import (
 
 type testService struct {
 	id *adapters.NodeId
+
+	// state stores []byte used to test creating and loading snapshots
+	state atomic.Value
 }
 
-func newTestService(id *adapters.NodeId) node.Service {
+func newTestService(id *adapters.NodeId, snapshot []byte) node.Service {
-	return &testService{id}
+	svc := &testService{id: id}
+	svc.state.Store(snapshot)
+	return svc
 }
 
 func (t *testService) Protocols() []p2p.Protocol {
@@ -35,7 +42,7 @@ func (t *testService) APIs() []rpc.API {
 	return []rpc.API{{
 		Namespace: "test",
 		Version:   "1.0",
-		Service:   &TestAPI{},
+		Service:   &TestAPI{state: &t.state},
 	}}
 }
 
@@ -56,9 +63,14 @@ func (t *testService) Run(_ *p2p.Peer, rw p2p.MsgReadWriter) error {
 	}
 }
 
+func (t *testService) Snapshot() ([]byte, error) {
+	return t.state.Load().([]byte), nil
+}
+
 // TestAPI provides a simple API to get and increment a counter and to
 // subscribe to increment events
 type TestAPI struct {
+	state   *atomic.Value
 	counter int64
 	feed    event.Feed
 }
@@ -72,6 +84,14 @@ func (t *TestAPI) Add(delta int64) {
 	t.feed.Send(delta)
 }
 
+func (t *TestAPI) GetState() []byte {
+	return t.state.Load().([]byte)
+}
+
+func (t *TestAPI) SetState(state []byte) {
+	t.state.Store(state)
+}
+
 func (t *TestAPI) Events(ctx context.Context) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)
 	if !supported {
@@ -106,14 +126,17 @@ var testServices = adapters.Services{
 	"test": newTestService,
 }
 
+func testHTTPServer(t *testing.T) *httptest.Server {
+	return httptest.NewServer(NewServer(&ServerConfig{
+		NewAdapter: func() adapters.NodeAdapter { return adapters.NewSimAdapter(testServices) },
+	}))
+}
+
 // TestHTTPNetwork tests creating and interacting with a simulation
 // network using the HTTP API
 func TestHTTPNetwork(t *testing.T) {
 	// start the server
-	srv := NewServer(&ServerConfig{
+	s := testHTTPServer(t)
-		Adapter: adapters.NewSimAdapter(testServices),
-	})
-	s := httptest.NewServer(srv)
 	defer s.Close()
 
 	// create a network
@@ -203,42 +226,55 @@ func TestHTTPNetwork(t *testing.T) {
 	}
 
 	// check we got all the events
-	nodeEvent := func(id string, up bool) *Event {
+	x := &expectEvents{t, events, sub}
-		return &Event{
+	x.expect(
-			Type: EventTypeNode,
+		x.nodeEvent(nodeIDs[0], false),
-			Node: &Node{
+		x.nodeEvent(nodeIDs[1], false),
-				Config: &adapters.NodeConfig{
+		x.nodeEvent(nodeIDs[0], true),
-					Id: adapters.NewNodeIdFromHex(id),
+		x.nodeEvent(nodeIDs[1], true),
-				},
+		x.connEvent(nodeIDs[0], nodeIDs[1], false),
-				Up: up,
+		x.connEvent(nodeIDs[0], nodeIDs[1], true),
+	)
+}
+
+type expectEvents struct {
+	*testing.T
+
+	events chan *Event
+	sub    event.Subscription
+}
+
+func (t *expectEvents) nodeEvent(id string, up bool) *Event {
+	return &Event{
+		Type: EventTypeNode,
+		Node: &Node{
+			Config: &adapters.NodeConfig{
+				Id: adapters.NewNodeIdFromHex(id),
 			},
-		}
+			Up: up,
+		},
 	}
-	connEvent := func(one, other string, up bool) *Event {
+}
-		return &Event{
+
-			Type: EventTypeConn,
+func (t *expectEvents) connEvent(one, other string, up bool) *Event {
-			Conn: &Conn{
+	return &Event{
-				One:   adapters.NewNodeIdFromHex(one),
+		Type: EventTypeConn,
-				Other: adapters.NewNodeIdFromHex(other),
+		Conn: &Conn{
-				Up:    up,
+			One:   adapters.NewNodeIdFromHex(one),
-			},
+			Other: adapters.NewNodeIdFromHex(other),
-		}
+			Up:    up,
-	}
+		},
-	expectedEvents := []*Event{
-		nodeEvent(nodeIDs[0], false),
-		nodeEvent(nodeIDs[1], false),
-		nodeEvent(nodeIDs[0], true),
-		nodeEvent(nodeIDs[1], true),
-		connEvent(nodeIDs[0], nodeIDs[1], false),
-		connEvent(nodeIDs[0], nodeIDs[1], true),
 	}
+}
+
+func (t *expectEvents) expect(events ...*Event) {
 	timeout := time.After(10 * time.Second)
-	for i := 0; i < len(expectedEvents); i++ {
+	for i := 0; i < len(events); i++ {
 		select {
-		case event := <-events:
+		case event := <-t.events:
 			t.Logf("received %s event: %s", event.Type, event)
 
-			expected := expectedEvents[i]
+			expected := events[i]
 			if event.Type != expected.Type {
 				t.Fatalf("expected event %d to have type %q, got %q", i, expected.Type, event.Type)
 			}
@@ -260,6 +296,12 @@ func TestHTTPNetwork(t *testing.T) {
 				if event.Conn == nil {
 					t.Fatal("expected event.Conn to be set")
 				}
+				if event.Conn.One == nil {
+					t.Fatal("expected event.Conn.One to be set")
+				}
+				if event.Conn.Other == nil {
+					t.Fatal("expected event.Conn.Other to be set")
+				}
 				if event.Conn.One.NodeID != expected.Conn.One.NodeID {
 					t.Fatalf("expected conn event %d to have one=%q, got one=%q", i, expected.Conn.One.Label(), event.Conn.One.Label())
 				}
@@ -271,7 +313,7 @@ func TestHTTPNetwork(t *testing.T) {
 				}
 			}
 
-		case err := <-sub.Err():
+		case err := <-t.sub.Err():
 			t.Fatalf("network stream closed unexpectedly: %s", err)
 
 		case <-timeout:
@@ -283,10 +325,7 @@ func TestHTTPNetwork(t *testing.T) {
 // TestHTTPNodeRPC tests calling RPC methods on nodes via the HTTP API
 func TestHTTPNodeRPC(t *testing.T) {
 	// start the server
-	srv := NewServer(&ServerConfig{
+	s := testHTTPServer(t)
-		Adapter: adapters.NewSimAdapter(testServices),
-	})
-	s := httptest.NewServer(srv)
 	defer s.Close()
 
 	// start a node in a network
@@ -345,3 +384,129 @@ func TestHTTPNodeRPC(t *testing.T) {
 		t.Fatal(ctx.Err())
 	}
 }
+
+// TestHTTPSnapshot tests creating and loading network snapshots
+func TestHTTPSnapshot(t *testing.T) {
+	// start the server
+	s := testHTTPServer(t)
+	defer s.Close()
+
+	// create a two-node network
+	client := NewClient(s.URL)
+	network, err := client.CreateNetwork(&NetworkConfig{DefaultService: "test"})
+	if err != nil {
+		t.Fatalf("error creating network: %s", err)
+	}
+	nodeCount := 2
+	nodes := make([]*p2p.NodeInfo, nodeCount)
+	for i := 0; i < nodeCount; i++ {
+		node, err := client.CreateNode(network.Id, &adapters.NodeConfig{})
+		if err != nil {
+			t.Fatalf("error creating node: %s", err)
+		}
+		if err := client.StartNode(network.Id, node.ID); err != nil {
+			t.Fatalf("error starting node: %s", err)
+		}
+		nodes[i] = node
+	}
+	if err := client.ConnectNode(network.Id, nodes[0].ID, nodes[1].ID); err != nil {
+		t.Fatalf("error connecting nodes: %s", err)
+	}
+
+	// store some state in the test services
+	states := make([]string, nodeCount)
+	for i, node := range nodes {
+		rpc, err := client.RPCClient(context.Background(), network.Id, node.ID)
+		if err != nil {
+			t.Fatalf("error getting RPC client: %s", err)
+		}
+		defer rpc.Close()
+		state := fmt.Sprintf("%x", rand.Int())
+		if err := rpc.Call(nil, "test_setState", []byte(state)); err != nil {
+			t.Fatalf("error setting service state: %s", err)
+		}
+		states[i] = state
+	}
+
+	// create a snapshot
+	snap, err := client.CreateSnapshot(network.Id)
+	if err != nil {
+		t.Fatalf("error creating snapshot: %s", err)
+	}
+	for i, state := range states {
+		if string(snap.Nodes[i].Snapshot) != state {
+			t.Fatalf("expected snapshot state %q, got %q", state, snap.Nodes[i].Snapshot)
+		}
+	}
+
+	// create another network
+	network, err = client.CreateNetwork(&NetworkConfig{DefaultService: "test"})
+	if err != nil {
+		t.Fatalf("error creating network: %s", err)
+	}
+
+	// subscribe to events so we can check them later
+	events := make(chan *Event, 100)
+	sub, err := client.SubscribeNetwork(network.Id, events)
+	if err != nil {
+		t.Fatalf("error subscribing to network events: %s", err)
+	}
+	defer sub.Unsubscribe()
+
+	// load the snapshot
+	if err := client.LoadSnapshot(network.Id, snap); err != nil {
+		t.Fatalf("error loading snapshot: %s", err)
+	}
+
+	// check the nodes and connection exists
+	net, err := client.GetNetwork(network.Id)
+	if err != nil {
+		t.Fatalf("error getting network: %s", err)
+	}
+	if len(net.Nodes) != nodeCount {
+		t.Fatalf("expected network to have %d nodes, got %d", nodeCount, len(net.Nodes))
+	}
+	for i, node := range nodes {
+		id := net.Nodes[i].ID().String()
+		if id != node.ID {
+			t.Fatalf("expected node %d to have ID %s, got %s", i, node.ID, id)
+		}
+	}
+	if len(net.Conns) != 1 {
+		t.Fatalf("expected network to have 1 connection, got %d", len(net.Conns))
+	}
+	conn := net.Conns[0]
+	if conn.One.String() != nodes[0].ID {
+		t.Fatalf("expected connection to have one=%q, got one=%q", nodes[0].ID, conn.One)
+	}
+	if conn.Other.String() != nodes[1].ID {
+		t.Fatalf("expected connection to have other=%q, got other=%q", nodes[1].ID, conn.Other)
+	}
+
+	// check the node states were restored
+	for i, node := range nodes {
+		rpc, err := client.RPCClient(context.Background(), network.Id, node.ID)
+		if err != nil {
+			t.Fatalf("error getting RPC client: %s", err)
+		}
+		defer rpc.Close()
+		var state []byte
+		if err := rpc.Call(&state, "test_getState"); err != nil {
+			t.Fatalf("error getting service state: %s", err)
+		}
+		if string(state) != states[i] {
+			t.Fatalf("expected snapshot state %q, got %q", states[i], state)
+		}
+	}
+
+	// check we got all the events
+	x := &expectEvents{t, events, sub}
+	x.expect(
+		x.nodeEvent(nodes[0].ID, false),
+		x.nodeEvent(nodes[0].ID, true),
+		x.nodeEvent(nodes[1].ID, false),
+		x.nodeEvent(nodes[1].ID, true),
+		x.connEvent(nodes[0].ID, nodes[1].ID, false),
+		x.connEvent(nodes[0].ID, nodes[1].ID, true),
+	)
+}

p2p/simulations/journal.go

@@ -1,235 +0,0 @@
-package simulations
-
-import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/ethereum/go-ethereum/event"
-	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
-)
-
-// Journal is an instance of a guaranteed no-loss subscription to network related events
-// (using event.TypeMux). Network components POST events to the TypeMux, which then is
-// read by the journal. Each journal belongs to a subscription.
-type Journal struct {
-	Id     string   `json:"id"`
-	Events []*Event `json:"events"`
-
-	lock    sync.Mutex
-	counter int
-	cursor  int
-	quitc   chan bool
-}
-
-// NewJournal constructor
-// Journal can get input events from subscriptions, add event logs
-// or scheduled replay of events from another journal
-//
-// see the Read and TimedRead iterators for use
-// the Journal is safe for concurrent reads and writes
-func NewJournal() *Journal {
-	return &Journal{quitc: make(chan bool)}
-}
-
-// Subscribe subscribes to an event.Feed and launches a gorourine that appends
-// any new event to the event log used for journalling history of a network
-// the goroutine terminates when the journal is closed
-func (self *Journal) Subscribe(feed *event.Feed) {
-	log.Info("subscribe")
-	events := make(chan *Event)
-	sub := feed.Subscribe(events)
-	go func() {
-		defer sub.Unsubscribe()
-		for {
-			select {
-			case event := <-events:
-				self.append(event)
-			case <-self.quitc:
-				return
-			}
-		}
-	}()
-}
-
-// AddJournal appends the event log of another journal to the receiver's one
-func (self *Journal) AddJournal(j *Journal) {
-	self.append(j.Events...)
-}
-
-// NewJournalFromJSON decodes a JSON serialised events log
-// into a journal struct
-// used to replay recorded history
-func NewJournalFromJSON(b []byte) (*Journal, error) {
-	self := NewJournal()
-	err := json.Unmarshal(b, self)
-	if err != nil {
-		return nil, err
-	}
-	return self, nil
-}
-
-// Replay replays the events of another journal preserving (relative) timing of events
-// params:
-// * acc: using acceleration factor acc
-// * journal: journal to use
-// * feed: where to post the replayed events
-func Replay(acc float64, j *Journal, feed *event.Feed) {
-	f := func(event *Event) bool {
-		// reposts the data with the eventer (the data receives a new timestamp)
-		event.Time = time.Now()
-		feed.Send(event)
-		return true
-	}
-	j.TimedRead(acc, f)
-}
-
-// Snapshot creates a snapshot out of the journal
-// this is simply done by reading the event log backwards and mark the last action
-// on a node/connection ignoring all earlier mentions
-// TODO: implmented
-func Snapshot(conf *SnapshotConfig, j *Journal) (*Journal, error) {
-	return nil, fmt.Errorf("snapshot not implemented")
-}
-
-func (self *Journal) Close() {
-	close(self.quitc)
-}
-
-func (self *Journal) append(events ...*Event) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	self.Events = append(self.Events, events...)
-	self.counter++
-}
-
-func (self *Journal) NewEntries() int {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	return self.counter - self.cursor
-}
-
-func (self *Journal) WaitEntries(n int) {
-	for self.NewEntries() < n {
-		time.Sleep(10 * time.Millisecond)
-	}
-}
-
-func (self *Journal) Read(f func(*Event) bool) (read int) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	ok := true
-	for self.cursor < len(self.Events) && ok {
-		read++
-		ok = f(self.Events[self.cursor])
-		self.cursor++
-		select {
-		case <-self.quitc:
-			break
-		default:
-		}
-	}
-	self.reset(self.cursor)
-	return read
-}
-
-// TimedRead reads the events but blocks for intervals that correspond to
-// the original time intervals,
-// NOTE: the events' timestamps are supposed to be strictly ordered otherwise
-// the call panics.
-// acc is an acceleration factor
-func (self *Journal) TimedRead(acc float64, f func(*Event) bool) (read int) {
-	var lastEvent time.Time
-	timer := time.NewTimer(0)
-	var event *Event
-	h := func(e *Event) bool {
-		// wait for the interval time passes event time
-		if e.Time.Before(lastEvent) {
-			panic("events not ordered")
-		}
-		interval := e.Time.Sub(lastEvent)
-		log.Trace(fmt.Sprintf("reset timer to interval %v", interval))
-		timer.Reset(time.Duration(acc) * interval)
-		lastEvent = e.Time
-		event = e
-		return false
-	}
-	var n int
-	for {
-		// Read blocks for the iteration. need to read one event at a time so that
-		// waiting for the timer to go off does not block concurrent access to the journal
-		n = self.Read(h)
-		if read > 0 && n > 0 {
-			select {
-			case <-self.quitc:
-				break
-			case <-timer.C:
-			}
-		}
-		read += n
-		if n == 0 || !f(event) {
-			log.Trace(fmt.Sprintf("timed read ends (read %v entries)", read))
-			break
-		}
-	}
-	return read
-}
-
-func (self *Journal) Reset(n int) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	self.reset(n)
-}
-
-func (self *Journal) reset(n int) {
-	length := len(self.Events)
-	if length == 0 {
-		return
-	}
-	if n >= length-1 {
-		n = length - 1
-	}
-	log.Trace(fmt.Sprintf("cursor reset from %v to %v/%v (%v)", self.cursor, n, len(self.Events), self.counter))
-	self.Events = self.Events[self.cursor:]
-	self.cursor = 0
-}
-
-func (self *Journal) Counter() int {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	return self.counter
-}
-
-// type History()
-
-func (self *Journal) Cursor() int {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	return self.cursor
-}
-
-type SnapshotConfig struct {
-	Id string
-}
-
-type JournalPlayConfig struct {
-	Id      string
-	SpeedUp float64
-	Journal *Journal
-	Events  []string
-}
-
-func ConnLabel(source, target *adapters.NodeId) string {
-	var first, second *adapters.NodeId
-	if bytes.Compare(source.Bytes(), target.Bytes()) > 0 {
-		first = target
-		second = source
-	} else {
-		first = source
-		second = target
-	}
-	return fmt.Sprintf("%v-%v", first, second)
-}

p2p/simulations/journal_test.go

@@ -1,143 +0,0 @@
-package simulations
-
-import (
-	"encoding/json"
-	"io/ioutil"
-	"testing"
-	"time"
-
-	"github.com/ethereum/go-ethereum/event"
-	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
-)
-
-func testEvents(intervals ...int) (events []*Event) {
-	t := time.Now()
-	for _, interval := range intervals {
-		t = t.Add(time.Duration(interval) * time.Millisecond)
-		events = append(events, &Event{
-			Type: EventTypeNode,
-			Time: t,
-			Node: &Node{Up: false},
-		})
-	}
-	return events
-}
-
-func TestTimedRead(t *testing.T) {
-	j := NewJournal()
-	intervals := []int{100, 200, 300, 300, 100, 200}
-	j.Events = testEvents(intervals...)
-	var newTimes []time.Time
-	var i int
-	acc := 0.5
-	length := 4
-	f := func(event *Event) bool {
-		newTimes = append(newTimes, time.Now())
-		i++
-		return i <= length
-	}
-	start := time.Now()
-	read := j.TimedRead(acc, f)
-	if read != 5 {
-		t.Fatalf("incorrect number of events read: expected 5, got %v", read)
-	}
-	for i, ti := range newTimes {
-		expInt := time.Duration(acc*float64(intervals[i])) * time.Millisecond
-		gotInt := ti.Sub(start)
-		if gotInt-expInt > 1*time.Millisecond {
-			t.Fatalf("journal timed read incorrect interval: expected %v ,got %v", expInt, gotInt)
-		}
-		start = ti
-	}
-}
-
-func testIDs() (ids []*adapters.NodeId) {
-
-	keys := []string{
-		"aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80",
-		"f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3",
-	}
-	for _, key := range keys {
-		id := adapters.NewNodeIdFromHex(key)
-		ids = append(ids, id)
-	}
-	return ids
-}
-
-func testJournal(ids []*adapters.NodeId) *Journal {
-	eventer := &event.Feed{}
-	journal := NewJournal()
-	journal.Subscribe(eventer)
-	for _, id := range ids {
-		eventer.Send(&Event{
-			Type: EventTypeNode,
-			Node: &Node{Config: &adapters.NodeConfig{Id: id}, Up: true},
-		})
-	}
-	journal.WaitEntries(len(ids))
-	return journal
-}
-
-func TestSubscribe(t *testing.T) {
-	ids := testIDs()
-	journal := testJournal(ids)
-	for i, ev := range journal.Events {
-		id := ev.Node.ID()
-		if id != ids[i] {
-			t.Fatalf("incorrect id: expected %v, got %v", id, ids[i])
-		}
-	}
-}
-
-func loadTestJournal(t *testing.T) ([]byte, *Journal) {
-	b, err := ioutil.ReadFile("./testjournal.json")
-	if err != nil {
-		t.Fatalf("unexpected error reading test journal json: %v", err)
-	}
-	journal, err := NewJournalFromJSON(b)
-	if err != nil {
-		t.Fatalf("unexpected error decoding journal json: %v", err)
-	}
-	return b, journal
-}
-
-func TestLoadSave(t *testing.T) {
-	b, j := loadTestJournal(t)
-
-	jo, err := json.MarshalIndent(j, "", " ")
-	if err != nil {
-		t.Fatalf("unexpected error encoding journal for %v: %v", j, err)
-	}
-	expJSON := string(b)
-	gotJSON := string(jo)
-	if expJSON != gotJSON {
-		t.Fatalf("incorrect json for journal: expected %v, got %v", expJSON, gotJSON)
-	}
-}
-
-func TestReplay(t *testing.T) {
-	_, jo := loadTestJournal(t)
-	eventer := &event.Feed{}
-
-	journal := NewJournal()
-	journal.Subscribe(eventer)
-
-	Replay(0, jo, eventer)
-	for i, ev := range jo.Events {
-		exp := ev.String()
-		got := journal.Events[i].String()
-		if exp != got {
-			t.Fatalf("incorrent replayed journal entry at pos %v: expected %v, got %v", i, exp, got)
-		}
-	}
-	// ids := RandomNodeIds(7)
-	// ticker := time.NewTicker(1000 * time.Microsecond)
-	// go MockEvents(eventer, ids, ticker.C)
-	// journal.WaitEntries(20)
-	// // eventer.Stop()	// eventer = &event.TypeMux{}
-	// journal = NewJournal()
-	// journal.Subscribe(eventer, &Entry{})
-
-	// func TestReplay(t *testing.T) {
-	// }
-}

p2p/simulations/network.go

@@ -27,6 +27,7 @@
 package simulations
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"sync"
@@ -261,6 +262,10 @@ func (self *Conn) nodesUp() error {
 
 // Start(id) starts up the node (relevant only for instance with own p2p or remote)
 func (self *Network) Start(id *adapters.NodeId) error {
+	return self.startWithSnapshot(id, nil)
+}
+
+func (self *Network) startWithSnapshot(id *adapters.NodeId, snapshot []byte) error {
 	node := self.GetNode(id)
 	if node == nil {
 		return fmt.Errorf("node %v does not exist", id)
@@ -269,7 +274,7 @@ func (self *Network) Start(id *adapters.NodeId) error {
 		return fmt.Errorf("node %v already up", id)
 	}
 	log.Trace(fmt.Sprintf("starting node %v: %v using %v", id, node.Up, self.nodeAdapter.Name()))
-	if err := node.Start(); err != nil {
+	if err := node.Start(snapshot); err != nil {
 		return err
 	}
 	node.Up = true
@@ -571,3 +576,75 @@ func (self *Network) Shutdown() {
 		}
 	}
 }
+
+func ConnLabel(source, target *adapters.NodeId) string {
+	var first, second *adapters.NodeId
+	if bytes.Compare(source.Bytes(), target.Bytes()) > 0 {
+		first = target
+		second = source
+	} else {
+		first = source
+		second = target
+	}
+	return fmt.Sprintf("%v-%v", first, second)
+}
+
+// Snapshot represents the state of a network at a single point in time and can
+// be used to restore the state of a network
+type Snapshot struct {
+	Nodes []NodeSnapshot `json:"nodes,omitempty"`
+	Conns []Conn         `json:"conns,omitempty"`
+}
+
+// NodeSnapshot represents the state of a node in the network
+type NodeSnapshot struct {
+	Node
+
+	// Snapshot is arbitrary data gathered from calling node.Snapshot()
+	Snapshot []byte `json:"snapshot,omitempty"`
+}
+
+// Snapshot creates a network snapshot
+func (self *Network) Snapshot() (*Snapshot, error) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	snap := &Snapshot{
+		Nodes: make([]NodeSnapshot, len(self.Nodes)),
+		Conns: make([]Conn, len(self.Conns)),
+	}
+	for i, node := range self.Nodes {
+		snap.Nodes[i] = NodeSnapshot{Node: *node}
+		if !node.Up {
+			continue
+		}
+		snapshot, err := node.Snapshot()
+		if err != nil {
+			return nil, err
+		}
+		snap.Nodes[i].Snapshot = snapshot
+	}
+	for i, conn := range self.Conns {
+		snap.Conns[i] = *conn
+	}
+	return snap, nil
+}
+
+func (self *Network) Load(snap *Snapshot) error {
+	for _, node := range snap.Nodes {
+		if _, err := self.NewNodeWithConfig(node.Config); err != nil {
+			return err
+		}
+		if !node.Up {
+			continue
+		}
+		if err := self.startWithSnapshot(node.Config.Id, node.Snapshot); err != nil {
+			return err
+		}
+	}
+	for _, conn := range snap.Conns {
+		if err := self.Connect(conn.One, conn.Other); err != nil {
+			return err
+		}
+	}
+	return nil
+}

swarm/network/simulations/discovery/discovery_test.go

@@ -22,7 +22,7 @@ import (
 const serviceName = "discovery"
 
 var services = adapters.Services{
-	serviceName: func(id *adapters.NodeId) p2pnode.Service {
+	serviceName: func(id *adapters.NodeId, snapshot []byte) p2pnode.Service {
 		return newNode(id)
 	},
 }

swarm/network/simulations/overlay.go

@@ -55,7 +55,7 @@ func (self *SimNode) Stop() error {
 }
 
 // NewSimNode creates adapters for nodes in the simulation.
-func NewSimNode(id *adapters.NodeId) node.Service {
+func NewSimNode(id *adapters.NodeId, snapshot []byte) node.Service {
 	addr := network.NewPeerAddrFromNodeId(id)
 	kp := network.NewKadParams()
 
@@ -155,8 +155,8 @@ func main() {
 	adapters.RegisterServices(services)
 
 	config := &simulations.ServerConfig{
-		Adapter: adapters.NewSimAdapter(services),
+		NewAdapter: func() adapters.NodeAdapter { return adapters.NewSimAdapter(services) },
-		Mocker:  mocker,
+		Mocker:     mocker,
 	}
 
 	log.Info("starting simulation server on 0.0.0.0:8888...")