Merge remote-tracking branch 'gagliardetto/multiepoch' into multiepoch

cmd-rpc.go

@@ -142,7 +142,7 @@ func newCmd_rpc() *cli.Command {
 				klog.Infof("Found %d directories; will start watching them for changes ...", len(dirs))
 				spew.Dump(dirs)
 
-				ctx, cancel := context.WithCancel(context.Background())
+				ctx, cancel := context.WithCancel(c.Context)
 				defer cancel()
 
 				err = onFileChanged(ctx, dirs, func(event fsnotify.Event) {
@@ -234,7 +234,7 @@ func newCmd_rpc() *cli.Command {
 				}
 			}
 
-			return multi.ListenAndServe(listenOn, listenerConfig)
+			return multi.ListenAndServe(c.Context, listenOn, listenerConfig)
 		},
 	}
 }

multiepoch-getBlock.go

@@ -363,7 +363,9 @@ func (multi *MultiEpoch) handleGetBlock(ctx context.Context, conn *requestContex
 	{
 		// get parent slot
 		parentSlot := uint64(block.Meta.Parent_slot)
-		if parentSlot != 0 {
+		if parentSlot != 0 && CalcEpochForSlot(parentSlot) == epochNumber {
+			// NOTE: if the parent is in the same epoch, we can get it from the same epoch handler as the block;
+			// otherwise, we need to get it from the previous epoch (TODO: implement this)
 			parentBlock, err := epochHandler.GetBlock(WithSubrapghPrefetch(ctx, false), parentSlot)
 			if err != nil {
 				return &jsonrpc2.Error{
@@ -384,6 +386,8 @@ func (multi *MultiEpoch) handleGetBlock(ctx context.Context, conn *requestContex
 				parentEntryHash := solana.HashFromBytes(parentEntryNode.Hash)
 				blockResp.PreviousBlockhash = parentEntryHash.String()
 			}
+		} else {
+			klog.Infof("parent slot is in a different epoch, not implemented yet")
 		}
 	}
 	tim.time("get parent block")

multiepoch.go

@@ -175,7 +175,7 @@ func LoadProxyConfig(configFilepath string) (*ProxyConfig, error) {
 }
 
 // ListeAndServe starts listening on the configured address and serves the RPC API.
-func (m *MultiEpoch) ListenAndServe(listenOn string, lsConf *ListenerConfig) error {
+func (m *MultiEpoch) ListenAndServe(ctx context.Context, listenOn string, lsConf *ListenerConfig) error {
 	handler := newMultiEpochHandler(m, lsConf)
 	handler = fasthttp.CompressHandler(handler)
 
@@ -206,7 +206,21 @@ func (m *MultiEpoch) ListenAndServe(listenOn string, lsConf *ListenerConfig) err
 	}
 
 	klog.Infof("RPC server listening on %s", listenOn)
-	return fasthttp.ListenAndServe(listenOn, handler)
+
+	s := &fasthttp.Server{
+		Handler:            handler,
+		MaxRequestBodySize: 1024 * 1024,
+	}
+	go func() {
+		// listen for context cancellation
+		<-ctx.Done()
+		klog.Info("RPC server shutting down...")
+		defer klog.Info("RPC server shut down")
+		if err := s.ShutdownWithContext(ctx); err != nil {
+			klog.Errorf("Error while shutting down RPC server: %s", err)
+		}
+	}()
+	return s.ListenAndServe(listenOn)
 }
 
 func randomRequestID() string {