[In-memory fan-out] CLI parameters update (#10250)

* [In-memory fan-out] CLI parameters update
* Enforce consistency for experimental CLI parameters for append-only, mutable state cache and in-memory fan-out
* Set default in-memory fan-out buffer size to 10K.

CHANGELOG_BEGIN
CHANGELOG_END

* Addressed review comments

ledger/participant-integration-api/src/main/scala/platform/index/ReadOnlySqlLedger.scala

@@ -30,6 +30,7 @@ import com.daml.timer.RetryStrategy
 
 import scala.concurrent.duration._
 import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.util.Failure
 
 private[platform] object ReadOnlySqlLedger {
 
@@ -67,19 +68,23 @@ private[platform] object ReadOnlySqlLedger {
       } yield ledger
 
     private def ledgerOwner(ledgerDao: LedgerReadDao, ledgerId: LedgerId) =
-      if (enableMutableContractStateCache)
-        new ReadOnlySqlLedgerWithMutableCache.Owner(
-          ledgerDao,
-          enricher,
-          ledgerId,
-          metrics,
-          maxContractStateCacheSize,
-          maxContractKeyStateCacheSize,
-          maxTransactionsInMemoryFanOutBufferSize,
-          enableInMemoryFanOutForLedgerApi,
-          servicesExecutionContext = servicesExecutionContext,
-        )
-      else
+      if (enableMutableContractStateCache) {
+        if (!enableAppendOnlySchema) {
+          failAppendOnlyNotEnabled()
+        } else {
+          new ReadOnlySqlLedgerWithMutableCache.Owner(
+            ledgerDao,
+            enricher,
+            ledgerId,
+            metrics,
+            maxContractStateCacheSize,
+            maxContractKeyStateCacheSize,
+            maxTransactionsInMemoryFanOutBufferSize,
+            enableInMemoryFanOutForLedgerApi,
+            servicesExecutionContext = servicesExecutionContext,
+          )
+        }
+      } else
         new ReadOnlySqlLedgerWithTranslationCache.Owner(
           ledgerDao,
           ledgerId,
@@ -156,6 +161,15 @@ private[platform] object ReadOnlySqlLedger {
           Some(enricher),
         )
   }
+
+  private def failAppendOnlyNotEnabled() =
+    ResourceOwner.forTry(() =>
+      Failure[ReadOnlySqlLedger](
+        new IllegalArgumentException(
+          "Mutable contract state cache must be enabled in conjunction with append-only schema"
+        )
+      )
+    )
 }
 
 private[index] abstract class ReadOnlySqlLedger(

ledger/participant-state/kvutils/app/src/main/scala/com/daml/ledger/participant/state/kvutils/app/Config.scala

@@ -510,7 +510,7 @@ object Config {
           .optional()
           .hidden()
           .text(
-            "Experimental contract state cache for command execution. Should not be used in production."
+            "Experimental contract state cache for command execution. Must be enabled in conjunction with index-append-only-schema-unsafe. Should not be used in production."
           )
           .action((_, config) => config.copy(enableMutableContractStateCache = true))
 
@@ -518,9 +518,23 @@ object Config {
           .optional()
           .hidden()
           .text(
-            "Experimental buffer for Ledger API streaming queries. Should not be used in production."
+            "Experimental buffer for Ledger API streaming queries. Must be enabled in conjunction with index-append-only-schema-unsafe and mutable-contract-state-cache-unsafe. Should not be used in production."
           )
           .action((_, config) => config.copy(enableInMemoryFanOutForLedgerApi = true))
+
+        checkConfig(config =>
+          if (config.enableMutableContractStateCache && !config.enableAppendOnlySchema)
+            failure(
+              "mutable-contract-state-cache-unsafe must be enabled in conjunction with index-append-only-schema-unsafe."
+            )
+          else if (
+            config.enableInMemoryFanOutForLedgerApi && !(config.enableMutableContractStateCache && config.enableAppendOnlySchema)
+          )
+            failure(
+              "buffered-ledger-api-streams-unsafe must be enabled in conjunction with index-append-only-schema-unsafe and mutable-contract-state-cache-unsafe."
+            )
+          else success
+        )
       }
     extraOptions(parser)
     parser

ledger/participant-state/kvutils/app/src/main/scala/com/daml/ledger/participant/state/kvutils/app/ParticipantConfig.scala

@@ -43,5 +43,5 @@ object ParticipantConfig {
   val DefaultMaxContractStateCacheSize: Long = 100000L
   val DefaultMaxContractKeyStateCacheSize: Long = 100000L
 
-  val DefaultMaxTransactionsInMemoryFanOutBufferSize: Long = 1000L
+  val DefaultMaxTransactionsInMemoryFanOutBufferSize: Long = 10000L
 }