Merge branch 'main' into cardano

Author: SupernaviX

Dockerfile

@@ -75,7 +75,7 @@ ARG UI_TAG
 ARG UI_RELEASE
 RUN apk add --update --no-cache \
   sqlite=3.44.2-r0 \
-  postgresql16-client=16.6-r0 \
+  postgresql16-client=16.8-r0 \
   curl=8.12.1-r0 \
   jq=1.7.1-r0
 WORKDIR /firefly

doc-site/docs/reference/config.md

@@ -407,30 +407,70 @@ title: Configuration Reference
 
 ## metrics
 
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|address|Deprecated - use monitoring.address instead|`int`|`127.0.0.1`
+|enabled|Deprecated - use monitoring.enabled instead|`boolean`|`true`
+|path|Deprecated - use monitoring.metricsPath instead|`string`|`/metrics`
+|port|Deprecated - use monitoring.port instead|`int`|`6000`
+|publicURL|Deprecated - use monitoring.publicURL instead|URL `string`|`<nil>`
+|readTimeout|Deprecated - use monitoring.readTimeout instead|[`time.Duration`](https://pkg.go.dev/time#Duration)|`15s`
+|shutdownTimeout|The maximum amount of time to wait for any open HTTP requests to finish before shutting down the HTTP server|[`time.Duration`](https://pkg.go.dev/time#Duration)|`10s`
+|writeTimeout|Deprecated - use monitoring.writeTimeout instead|[`time.Duration`](https://pkg.go.dev/time#Duration)|`15s`
+
+## metrics.auth
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|type|The auth plugin to use for server side authentication of requests|`string`|`<nil>`
+
+## metrics.auth.basic
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|passwordfile|The path to a .htpasswd file to use for authenticating requests. Passwords should be hashed with bcrypt.|`string`|`<nil>`
+
+## metrics.tls
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|ca|The TLS certificate authority in PEM format (this option is ignored if caFile is also set)|`string`|`<nil>`
+|caFile|The path to the CA file for TLS on this API|`string`|`<nil>`
+|cert|The TLS certificate in PEM format (this option is ignored if certFile is also set)|`string`|`<nil>`
+|certFile|The path to the certificate file for TLS on this API|`string`|`<nil>`
+|clientAuth|Enables or disables client auth for TLS on this API|`string`|`<nil>`
+|enabled|Enables or disables TLS on this API|`boolean`|`false`
+|insecureSkipHostVerify|When to true in unit test development environments to disable TLS verification. Use with extreme caution|`boolean`|`<nil>`
+|key|The TLS certificate key in PEM format (this option is ignored if keyFile is also set)|`string`|`<nil>`
+|keyFile|The path to the private key file for TLS on this API|`string`|`<nil>`
+|requiredDNAttributes|A set of required subject DN attributes. Each entry is a regular expression, and the subject certificate must have a matching attribute of the specified type (CN, C, O, OU, ST, L, STREET, POSTALCODE, SERIALNUMBER are valid attributes)|`map[string]string`|`<nil>`
+
+## monitoring
+
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|
 |address|The IP address on which the metrics HTTP API should listen|`int`|`127.0.0.1`
 |enabled|Enables the metrics API|`boolean`|`true`
-|path|The path from which to serve the Prometheus metrics|`string`|`/metrics`
+|metricsPath|The path from which to serve the Prometheus metrics|`string`|`/metrics`
 |port|The port on which the metrics HTTP API should listen|`int`|`6000`
 |publicURL|The fully qualified public URL for the metrics API. This is used for building URLs in HTTP responses and in OpenAPI Spec generation|URL `string`|`<nil>`
 |readTimeout|The maximum time to wait when reading from an HTTP connection|[`time.Duration`](https://pkg.go.dev/time#Duration)|`15s`
 |shutdownTimeout|The maximum amount of time to wait for any open HTTP requests to finish before shutting down the HTTP server|[`time.Duration`](https://pkg.go.dev/time#Duration)|`10s`
 |writeTimeout|The maximum time to wait when writing to an HTTP connection|[`time.Duration`](https://pkg.go.dev/time#Duration)|`15s`
 
-## metrics.auth
+## monitoring.auth
 
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|
 |type|The auth plugin to use for server side authentication of requests|`string`|`<nil>`
 
-## metrics.auth.basic
+## monitoring.auth.basic
 
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|
 |passwordfile|The path to a .htpasswd file to use for authenticating requests. Passwords should be hashed with bcrypt.|`string`|`<nil>`
 
-## metrics.tls
+## monitoring.tls
 
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|

internal/apiserver/metrics_server.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -21,11 +21,17 @@ import (
 )
 
 const (
-	MetricsEnabled = "enabled"
-	MetricsPath    = "path"
+	Enabled               = "enabled"
+	DeprecatedMetricsPath = "path"
+	MetricsPath           = "metricsPath"
 )
 
-func initMetricsConfig(config config.Section) {
-	config.AddKnownKey(MetricsEnabled, true)
+func initDeprecatedMetricsConfig(config config.Section) {
+	config.AddKnownKey(Enabled, true)
+	config.AddKnownKey(DeprecatedMetricsPath, "/metrics")
+}
+
+func initMonitoringConfig(config config.Section) {
+	config.AddKnownKey(Enabled, true)
 	config.AddKnownKey(MetricsPath, "/metrics")
 }

internal/apiserver/server.go

@@ -1,4 +1,4 @@
-// Copyright © 2024 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -42,10 +42,11 @@ import (
 )
 
 var (
-	spiConfig     = config.RootSection("spi")
-	apiConfig     = config.RootSection("http")
-	metricsConfig = config.RootSection("metrics")
-	corsConfig    = config.RootSection("cors")
+	spiConfig               = config.RootSection("spi")
+	apiConfig               = config.RootSection("http")
+	deprecatedMetricsConfig = config.RootSection("metrics")
+	monitoringConfig        = config.RootSection("monitoring")
+	corsConfig              = config.RootSection("cors")
 )
 
 // Server is the external interface for the API Server
@@ -55,31 +56,35 @@ type Server interface {
 
 type apiServer struct {
 	// Defaults set with config
-	apiTimeout             time.Duration
-	apiMaxTimeout          time.Duration
-	metricsEnabled         bool
-	ffiSwaggerGen          FFISwaggerGen
-	apiPublicURL           string
-	dynamicPublicURLHeader string
-	defaultNamespace       string
+	apiTimeout               time.Duration
+	apiMaxTimeout            time.Duration
+	deprecatedMetricsEnabled bool
+	monitoringEnabled        bool
+	ffiSwaggerGen            FFISwaggerGen
+	apiPublicURL             string
+	dynamicPublicURLHeader   string
+	defaultNamespace         string
 }
 
 func InitConfig() {
 	httpserver.InitHTTPConfig(apiConfig, 5000)
 	httpserver.InitHTTPConfig(spiConfig, 5001)
-	httpserver.InitHTTPConfig(metricsConfig, 6000)
+	httpserver.InitHTTPConfig(deprecatedMetricsConfig, 6000)
+	httpserver.InitHTTPConfig(monitoringConfig, 6000)
 	httpserver.InitCORSConfig(corsConfig)
-	initMetricsConfig(metricsConfig)
+	initDeprecatedMetricsConfig(deprecatedMetricsConfig)
+	initMonitoringConfig(monitoringConfig)
 }
 
 func NewAPIServer() Server {
 	as := &apiServer{
-		apiTimeout:             config.GetDuration(coreconfig.APIRequestTimeout),
-		apiMaxTimeout:          config.GetDuration(coreconfig.APIRequestMaxTimeout),
-		dynamicPublicURLHeader: config.GetString(coreconfig.APIDynamicPublicURLHeader),
-		defaultNamespace:       config.GetString(coreconfig.NamespacesDefault),
-		metricsEnabled:         config.GetBool(coreconfig.MetricsEnabled),
-		ffiSwaggerGen:          &ffiSwaggerGen{},
+		apiTimeout:               config.GetDuration(coreconfig.APIRequestTimeout),
+		apiMaxTimeout:            config.GetDuration(coreconfig.APIRequestMaxTimeout),
+		dynamicPublicURLHeader:   config.GetString(coreconfig.APIDynamicPublicURLHeader),
+		defaultNamespace:         config.GetString(coreconfig.NamespacesDefault),
+		deprecatedMetricsEnabled: config.GetBool(coreconfig.DeprecatedMetricsEnabled),
+		monitoringEnabled:        config.GetBool(coreconfig.MonitoringEnabled),
+		ffiSwaggerGen:            &ffiSwaggerGen{},
 	}
 	as.apiPublicURL = as.getPublicURL(apiConfig, "")
 	return as
@@ -110,15 +115,21 @@ func (as *apiServer) Serve(ctx context.Context, mgr namespace.Manager) (err erro
 	} else if config.GetBool(coreconfig.LegacyAdminEnabled) {
 		log.L(ctx).Warnf("Your config includes an 'admin' section, which should be renamed to 'spi' - SPI server will not be enabled until this is corrected")
 	}
+	serverName := "metrics"
+	mConfig := deprecatedMetricsConfig
+	if as.monitoringEnabled {
+		serverName = "monitoring"
+		mConfig = monitoringConfig
+	}
 
-	if as.metricsEnabled {
-		metricsHTTPServer, err := httpserver.NewHTTPServer(ctx, "metrics", as.createMetricsMuxRouter(), metricsErrChan, metricsConfig, corsConfig, &httpserver.ServerOptions{
+	if as.deprecatedMetricsEnabled || as.monitoringEnabled {
+		monitoringServer, err := httpserver.NewHTTPServer(ctx, serverName, as.createMonitoringMuxRouter(), metricsErrChan, mConfig, corsConfig, &httpserver.ServerOptions{
 			MaximumRequestTimeout: as.apiMaxTimeout,
 		})
 		if err != nil {
 			return err
 		}
-		go metricsHTTPServer.ServeHTTP(ctx)
+		go monitoringServer.ServeHTTP(ctx)
 	}
 
 	return as.waitForServerStop(httpErrChan, spiErrChan, metricsErrChan)
@@ -345,7 +356,7 @@ func (as *apiServer) createMuxRouter(ctx context.Context, mgr namespace.Manager)
 	r := mux.NewRouter()
 	hf := as.handlerFactory()
 
-	if as.metricsEnabled {
+	if as.deprecatedMetricsEnabled || as.monitoringEnabled {
 		r.Use(metrics.GetRestServerInstrumentation().Middleware)
 	}
 
@@ -433,7 +444,7 @@ func (as *apiServer) spiWSHandler(mgr namespace.Manager) http.HandlerFunc {
 
 func (as *apiServer) createAdminMuxRouter(mgr namespace.Manager) *mux.Router {
 	r := mux.NewRouter()
-	if as.metricsEnabled {
+	if as.deprecatedMetricsEnabled || as.monitoringEnabled {
 		r.Use(metrics.GetAdminServerInstrumentation().Middleware)
 	}
 	hf := as.handlerFactory()
@@ -468,12 +479,20 @@ func (as *apiServer) createAdminMuxRouter(mgr namespace.Manager) *mux.Router {
 	return r
 }
 
-func (as *apiServer) createMetricsMuxRouter() *mux.Router {
+func (as *apiServer) createMonitoringMuxRouter() *mux.Router {
 	r := mux.NewRouter()
-
-	r.Path(config.GetString(coreconfig.MetricsPath)).Handler(promhttp.InstrumentMetricHandler(metrics.Registry(),
+	metricsPath := config.GetString(coreconfig.DeprecatedMetricsPath)
+	if as.monitoringEnabled {
+		metricsPath = config.GetString(coreconfig.MonitoringMetricsPath)
+	}
+	r.Path(metricsPath).Handler(promhttp.InstrumentMetricHandler(metrics.Registry(),
 		promhttp.HandlerFor(metrics.Registry(), promhttp.HandlerOpts{})))
-
+	hf := ffapi.HandlerFactory{}
+	r.HandleFunc("/livez", hf.APIWrapper(func(res http.ResponseWriter, req *http.Request) (status int, err error) {
+		// a simple liveness check
+		return http.StatusOK, nil
+	}))
+	r.NotFoundHandler = hf.APIWrapper(as.notFoundHandler)
 	return r
 }
 

internal/apiserver/server_test.go

@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -86,7 +86,7 @@ func TestStartStopServer(t *testing.T) {
 	InitConfig()
 	apiConfig.Set(httpserver.HTTPConfPort, 0)
 	spiConfig.Set(httpserver.HTTPConfPort, 0)
-	metricsConfig.Set(httpserver.HTTPConfPort, 0)
+	monitoringConfig.Set(httpserver.HTTPConfPort, 0)
 	config.Set(coreconfig.UIPath, "test")
 	config.Set(coreconfig.SPIEnabled, true)
 	ctx, cancel := context.WithCancel(context.Background())
@@ -105,7 +105,7 @@ func TestStartLegacyAdminConfig(t *testing.T) {
 	InitConfig()
 	apiConfig.Set(httpserver.HTTPConfPort, 0)
 	spiConfig.Set(httpserver.HTTPConfPort, 0)
-	metricsConfig.Set(httpserver.HTTPConfPort, 0)
+	monitoringConfig.Set(httpserver.HTTPConfPort, 0)
 	config.Set(coreconfig.UIPath, "test")
 	config.Set(coreconfig.LegacyAdminEnabled, true)
 	ctx, cancel := context.WithCancel(context.Background())
@@ -170,8 +170,8 @@ func TestStartMetricsFail(t *testing.T) {
 	coreconfig.Reset()
 	metrics.Clear()
 	InitConfig()
-	metricsConfig.Set(httpserver.HTTPConfAddress, "...://")
-	config.Set(coreconfig.MetricsEnabled, true)
+	monitoringConfig.Set(httpserver.HTTPConfAddress, "...://")
+	config.Set(coreconfig.MonitoringEnabled, true)
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel() // server will immediately shut down
 	as := NewAPIServer()
@@ -575,3 +575,17 @@ func TestContractAPIDefaultNS(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, 200, res.StatusCode())
 }
+
+func TestMonitoringServerRoutes(t *testing.T) {
+	_, _, as := newTestServer()
+	s := httptest.NewServer(as.createMonitoringMuxRouter())
+	defer s.Close()
+
+	res, err := http.Get(fmt.Sprintf("http://%s/livez", s.Listener.Addr()))
+	assert.NoError(t, err)
+	assert.Equal(t, 200, res.StatusCode)
+
+	res, err = http.Get(fmt.Sprintf("http://%s/metrics", s.Listener.Addr()))
+	assert.NoError(t, err)
+	assert.Equal(t, 200, res.StatusCode)
+}

internal/blockchain/common/common.go

@@ -1,4 +1,4 @@
-// Copyright © 2024 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -43,6 +43,12 @@ type BlockchainCallbacks interface {
 	SetHandler(namespace string, handler blockchain.Callbacks)
 	SetOperationalHandler(namespace string, handler core.OperationCallbacks)
 
+	// BulkOperationUpdates is a synchronous way to update multiple operations and will return when the updates have been committed to the database or there has been an error
+	// An insertion ordering guarantee is only provided when this code is called on a single goroutine inside of the connector.
+	// It is the responsibility of the connector code to allocate that routine, and ensure that there is only one.
+	// Note: onComplete at each update level is not called, as this is a bulk operation and should be reponsibility of the caller to manage if needed.
+	BulkOperationUpdates(ctx context.Context, namespace string, updates []*core.OperationUpdate) error
+
 	OperationUpdate(ctx context.Context, plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject)
 	// Common logic for parsing a BatchPinOrNetworkAction event, and if not discarded to add it to the by-namespace map
 	PrepareBatchPinOrNetworkAction(ctx context.Context, events EventsToDispatch, subInfo *SubscriptionInfo, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef, params *BatchPinParams)
@@ -64,6 +70,17 @@ type callbacks struct {
 	opHandlers map[string]core.OperationCallbacks
 }
 
+// BulkOperationUpdates implements BlockchainCallbacks.
+func (cb *callbacks) BulkOperationUpdates(ctx context.Context, namespace string, updates []*core.OperationUpdate) error {
+	if handler, ok := cb.opHandlers[namespace]; ok {
+		return handler.BulkOperationUpdates(ctx, updates)
+	}
+	// We don't want to error as it just means this update was not for this namespace
+	// This is unlikely to happen in practice, as the namespace is always passed in the operation handler
+	log.L(ctx).Errorf("No operation handler found for namespace '%s'", namespace)
+	return nil
+}
+
 type subscriptions struct {
 	subs map[string]*SubscriptionInfo
 }
@@ -153,13 +170,15 @@ func (cb *callbacks) SetOperationalHandler(namespace string, handler core.Operat
 func (cb *callbacks) OperationUpdate(ctx context.Context, plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
 	namespace, _, _ := core.ParseNamespacedOpID(ctx, nsOpID)
 	if handler, ok := cb.opHandlers[namespace]; ok {
-		handler.OperationUpdate(&core.OperationUpdate{
-			Plugin:         plugin.Name(),
-			NamespacedOpID: nsOpID,
-			Status:         status,
-			BlockchainTXID: blockchainTXID,
-			ErrorMessage:   errorMessage,
-			Output:         opOutput,
+		handler.OperationUpdate(&core.OperationUpdateAsync{
+			OperationUpdate: core.OperationUpdate{
+				Plugin:         plugin.Name(),
+				NamespacedOpID: nsOpID,
+				Status:         status,
+				BlockchainTXID: blockchainTXID,
+				ErrorMessage:   errorMessage,
+				Output:         opOutput,
+			},
 		})
 		return
 	}