Merge pull request #587 from formancehq/feat/parallel-bulk

feat(bulk): add parallel processing
formancehq · Nov 25, 2024 · 9d948b7 · 9d948b7
2 parents 26eaec1 + fb4c804
commit 9d948b7
Show file tree

Hide file tree

Showing 25 changed files with 440 additions and 245 deletions.
diff --git a/cmd/serve.go b/cmd/serve.go
@@ -41,6 +41,7 @@ const (
 	AutoUpgradeFlag            = "auto-upgrade"
 	ExperimentalFeaturesFlag   = "experimental-features"
 	BulkMaxSizeFlag            = "bulk-max-size"
+	BulkParallelFlag           = "bulk-parallel"
 	NumscriptInterpreterFlag   = "experimental-numscript-interpreter"
 )
 
@@ -67,6 +68,11 @@ func NewServeCommand() *cobra.Command {
 				return err
 			}
 
+			bulkParallel, err := cmd.Flags().GetInt(BulkParallelFlag)
+			if err != nil {
+				return err
+			}
+
 			options := []fx.Option{
 				fx.NopLogger,
 				otlp.FXModuleFromFlags(cmd),
@@ -90,9 +96,12 @@ func NewServeCommand() *cobra.Command {
 				bus.NewFxModule(),
 				ballast.Module(serveConfiguration.ballastSize),
 				api.Module(api.Config{
-					Version:     Version,
-					Debug:       service.IsDebug(cmd),
-					BulkMaxSize: bulkMaxSize,
+					Version: Version,
+					Debug:   service.IsDebug(cmd),
+					Bulk: api.BulkConfig{
+						MaxSize:  bulkMaxSize,
+						Parallel: bulkParallel,
+					},
 				}),
 				fx.Decorate(func(
 					params struct {
@@ -129,6 +138,7 @@ func NewServeCommand() *cobra.Command {
 	cmd.Flags().String(BindFlag, "0.0.0.0:3068", "API bind address")
 	cmd.Flags().Bool(ExperimentalFeaturesFlag, false, "Enable features configurability")
 	cmd.Flags().Int(BulkMaxSizeFlag, api.DefaultBulkMaxSize, "Bulk max size (default 100)")
+	cmd.Flags().Int(BulkParallelFlag, 10, "Bulk max parallelism")
 	cmd.Flags().Bool(NumscriptInterpreterFlag, false, "Enable experimental numscript rewrite")
 
 	service.AddFlags(cmd.Flags())

diff --git a/docs/api/README.md b/docs/api/README.md
@@ -500,6 +500,7 @@ Accept: application/json
 |ledger|path|string|true|Name of the ledger.|
 |continueOnFailure|query|boolean|false|Continue on failure|
 |atomic|query|boolean|false|Make bulk atomic|
+|parallel|query|boolean|false|Process bulk elements in parallel|
 |body|body|[V2Bulk](#schemav2bulk)|false|none|
 
 > Example responses

diff --git a/internal/api/module.go b/internal/api/module.go
@@ -4,16 +4,22 @@ import (
 	_ "embed"
 	"github.com/formancehq/go-libs/v2/auth"
 	"github.com/formancehq/go-libs/v2/health"
+	"github.com/formancehq/ledger/internal/controller/ledger"
 	"github.com/formancehq/ledger/internal/controller/system"
 	"github.com/go-chi/chi/v5"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/fx"
 )
 
+type BulkConfig struct {
+	MaxSize int
+	Parallel int
+}
+
 type Config struct {
 	Version     string
 	Debug       bool
-	BulkMaxSize int
+	Bulk BulkConfig
 }
 
 func Module(cfg Config) fx.Option {
@@ -29,7 +35,10 @@ func Module(cfg Config) fx.Option {
 				"develop",
 				cfg.Debug,
 				WithTracer(tracer.Tracer("api")),
-				WithBulkMaxSize(cfg.BulkMaxSize),
+				WithBulkMaxSize(cfg.Bulk.MaxSize),
+				WithBulkerFactory(ledger.NewDefaultBulkerFactory(
+					ledger.WithParallelism(cfg.Bulk.Parallel),
+				)),
 			)
 		}),
 		health.Module(),

diff --git a/internal/api/router.go b/internal/api/router.go
@@ -2,6 +2,7 @@ package api
 
 import (
 	"github.com/formancehq/go-libs/v2/api"
+	ledgercontroller "github.com/formancehq/ledger/internal/controller/ledger"
 	"github.com/formancehq/ledger/internal/controller/system"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
@@ -66,6 +67,7 @@ func NewRouter(
 		v2.WithTracer(routerOptions.tracer),
 		v2.WithMiddlewares(commonMiddlewares...),
 		v2.WithBulkMaxSize(routerOptions.bulkMaxSize),
+		v2.WithBulkerFactory(routerOptions.bulkerFactory),
 	)
 	mux.Handle("/v2*", http.StripPrefix("/v2", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		chi.RouteContext(r.Context()).Reset()
@@ -84,8 +86,9 @@ func NewRouter(
 }
 
 type routerOptions struct {
-	tracer      trace.Tracer
-	bulkMaxSize int
+	tracer        trace.Tracer
+	bulkMaxSize   int
+	bulkerFactory ledgercontroller.BulkerFactory
 }
 
 type RouterOption func(ro *routerOptions)
@@ -102,6 +105,12 @@ func WithBulkMaxSize(bulkMaxSize int) RouterOption {
 	}
 }
 
+func WithBulkerFactory(bf ledgercontroller.BulkerFactory) RouterOption {
+	return func(ro *routerOptions) {
+		ro.bulkerFactory = bf
+	}
+}
+
 var defaultRouterOptions = []RouterOption{
 	WithTracer(nooptracer.Tracer{}),
 	WithBulkMaxSize(DefaultBulkMaxSize),

diff --git a/internal/api/v2/controllers_bulk.go b/internal/api/v2/controllers_bulk.go
@@ -12,7 +12,7 @@ import (
 	ledgercontroller "github.com/formancehq/ledger/internal/controller/ledger"
 )
 
-func bulkHandler(bulkMaxSize int) http.HandlerFunc {
+func bulkHandler(bulkerFactory ledgercontroller.BulkerFactory, bulkMaxSize int) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		b := ledgercontroller.Bulk{}
 		if err := json.NewDecoder(r.Body).Decode(&b); err != nil {
@@ -28,11 +28,11 @@ func bulkHandler(bulkMaxSize int) http.HandlerFunc {
 		w.Header().Set("Content-Type", "application/json")
 
 		ledgerController := common.LedgerFromContext(r.Context())
-		bulker := ledgercontroller.NewBulker(ledgerController)
-		results, err := bulker.Run(r.Context(),
-			b,
-			api.QueryParamBool(r, "continueOnFailure"),
-			api.QueryParamBool(r, "atomic"),
+
+		results, err := bulkerFactory.CreateBulker(ledgerController).Run(r.Context(), b,
+			ledgercontroller.WithContinueOnFailure(api.QueryParamBool(r, "continueOnFailure")),
+			ledgercontroller.WithAtomic(api.QueryParamBool(r, "atomic")),
+			ledgercontroller.WithParallel(api.QueryParamBool(r, "parallel")),
 		)
 		if err != nil {
 			api.InternalServerError(w, r, err)

diff --git a/internal/api/v2/controllers_bulk_test.go b/internal/api/v2/controllers_bulk_test.go
@@ -265,6 +265,10 @@ func TestBulk(t *testing.T) {
 				ErrorCode:        api.ErrorInternal,
 				ErrorDescription: "unexpected error",
 				ResponseType:     "ERROR",
+			}, {
+				ErrorCode:        api.ErrorInternal,
+				ErrorDescription: "canceled",
+				ResponseType:     "ERROR",
 			}},
 			expectError: true,
 		},

diff --git a/internal/api/v2/routes.go b/internal/api/v2/routes.go
@@ -1,6 +1,7 @@
 package v2
 
 import (
+	"github.com/formancehq/ledger/internal/controller/ledger"
 	nooptracer "go.opentelemetry.io/otel/trace/noop"
 	"net/http"
 
@@ -52,7 +53,7 @@ func NewRouter(
 			router.With(common.LedgerMiddleware(systemController, func(r *http.Request) string {
 				return chi.URLParam(r, "ledger")
 			}, routerOptions.tracer, "/_info")).Group(func(router chi.Router) {
-				router.Post("/_bulk", bulkHandler(routerOptions.bulkMaxSize))
+				router.Post("/_bulk", bulkHandler(routerOptions.bulkerFactory, routerOptions.bulkMaxSize))
 
 				// LedgerController
 				router.Get("/_info", getLedgerInfo)
@@ -92,6 +93,7 @@ func NewRouter(
 type routerOptions struct {
 	tracer      trace.Tracer
 	middlewares []func(http.Handler) http.Handler
+	bulkerFactory ledger.BulkerFactory
 	bulkMaxSize int
 }
 
@@ -115,6 +117,13 @@ func WithBulkMaxSize(bulkMaxSize int) RouterOption {
 	}
 }
 
+func WithBulkerFactory(bulkerFactory ledger.BulkerFactory) RouterOption {
+	return func(ro *routerOptions) {
+		ro.bulkerFactory = bulkerFactory
+	}
+}
+
 var defaultRouterOptions = []RouterOption{
 	WithTracer(nooptracer.Tracer{}),
+	WithBulkerFactory(ledger.NewDefaultBulkerFactory()),
 }