Ensure that on first call to ReadService.stateUpdates beginOffset is … (#6103)

* Ensure that on first call to ReadService.stateUpdates beginOffset is None

Fixes #6023

CHANGELOG_BEGIN
CHANGELOG_END

Also restores correct advice in CONTRIBUTING.md

* Update with review comments

CONTRIBUTING.md

@@ -21,7 +21,7 @@ For Git commit messages, our principle is that `git log --pretty=oneline` should
   * *Improve explanation of …*
   * *Remove module X because it is not used.*
 * Commits should have a description that concisely explains the rationale and context for the change if that is not obvious.
-* The first line of the description should include a reference to the fixed issue `(#XXX)`.
+* Commit descriptions should include a `Fixes #XX` line indicating what GitHub issue number the commit fixes.
 * The git logs are not intended for user-facing change logs, but should be a useful reference when writing them.
 
 ## Pull request checklist
@@ -35,14 +35,14 @@ For Git commit messages, our principle is that `git log --pretty=oneline` should
 
 The following is an example of a well-formed commit, including the description (first line) and a body that includes changelog additions:
 
-      Introduced a new API for package management (#1311)
+      Fixes #1311
 
-      Also fixes a typo in the Scala bindings documentation
+      Also fixes a typo in the Scala bindings documentation.
 
       CHANGELOG_BEGIN
 
       - [Sandbox] Introduced a new API for package management.
-      See `#1311 <https://github.com/digital-asset/daml/issues/1311>`__.
+        See `#1311 <https://github.com/digital-asset/daml/issues/1311>`__.
 
       CHANGELOG_END
 

ledger/sandbox/src/main/scala/com/digitalasset/platform/indexer/JdbcIndexer.scala

@@ -109,7 +109,7 @@ final class JdbcIndexerFactory(
       ledgerDao: LedgerDao,
   ): Future[Unit] = {
     logger.info(s"Initializing ledger with ID: $providedLedgerId")
-    ledgerDao.initializeLedger(providedLedgerId, Offset.begin)
+    ledgerDao.initializeLedger(providedLedgerId)
   }
 }
 

ledger/sandbox/src/main/scala/com/digitalasset/platform/sandbox/stores/ledger/sql/SqlLedger.scala

@@ -443,7 +443,7 @@ private final class SqlLedgerFactory(ledgerDao: LedgerDao)(implicit logCtx: Logg
       _ <- if (initializationRequired) {
         logger.info(s"Initializing ledger with ID: $ledgerId")
         for {
-          _ <- ledgerDao.initializeLedger(ledgerId, Offset.begin)
+          _ <- ledgerDao.initializeLedger(ledgerId)
           _ <- initializeLedgerEntries(
             initialLedgerEntries,
             timeProvider,

ledger/sandbox/src/main/scala/com/digitalasset/platform/store/dao/JdbcLedgerDao.scala

@@ -110,10 +110,13 @@ private class JdbcLedgerDao(
 
   private val SQL_SELECT_LEDGER_END = SQL("select ledger_end from parameters")
 
+  /**
+    * Defaults to Offset.begin if ledger_end is unset
+    */
   override def lookupLedgerEnd(): Future[Offset] =
     dbDispatcher.executeSql(metrics.daml.index.db.getLedgerEnd) { implicit conn =>
       SQL_SELECT_LEDGER_END
-        .as(offset("ledger_end").single)
+        .as(offset("ledger_end").?.map(_.getOrElse(Offset.begin)).single)
     }
 
   private val SQL_SELECT_INITIAL_LEDGER_END = SQL("select ledger_end from parameters")
@@ -124,13 +127,12 @@ private class JdbcLedgerDao(
         .as(offset("ledger_end").?.single)
     }
 
-  private val SQL_INITIALIZE = SQL(
-    "insert into parameters(ledger_id, ledger_end) VALUES({LedgerId}, {LedgerEnd})")
+  private val SQL_INITIALIZE = SQL("insert into parameters(ledger_id) VALUES({LedgerId})")
 
-  override def initializeLedger(ledgerId: LedgerId, ledgerEnd: Offset): Future[Unit] =
+  override def initializeLedger(ledgerId: LedgerId): Future[Unit] =
     dbDispatcher.executeSql(metrics.daml.index.db.initializeLedgerParameters) { implicit conn =>
       val _ = SQL_INITIALIZE
-        .on("LedgerId" -> ledgerId.unwrap, "LedgerEnd" -> ledgerEnd)
+        .on("LedgerId" -> ledgerId.unwrap)
         .execute()
       ()
     }
@@ -139,7 +141,7 @@ private class JdbcLedgerDao(
   // and thus we need to make sure to only update the ledger end when the ledger entry we're committing
   // is advancing it.
   private val SQL_UPDATE_LEDGER_END = SQL(
-    "update parameters set ledger_end = {LedgerEnd} where ledger_end < {LedgerEnd}")
+    "update parameters set ledger_end = {LedgerEnd} where (ledger_end is null or ledger_end < {LedgerEnd})")
 
   private def updateLedgerEnd(ledgerEnd: Offset)(implicit conn: Connection): Unit = {
     SQL_UPDATE_LEDGER_END
@@ -166,15 +168,15 @@ private class JdbcLedgerDao(
   }
 
   private val currentConfigurationParser: ResultSetParser[Option[(Offset, Configuration)]] =
-    (offset("ledger_end") ~
+    (offset("ledger_end").? ~
       byteArray("configuration").? map flatten).single
       .map {
-        case (_, None) => None
-        case (offset, Some(configBytes)) =>
+        case (Some(offset), Some(configBytes)) =>
           Configuration
             .decode(configBytes)
             .toOption
             .map(config => offset -> config)
+        case _ => None
       }
 
   private def selectLedgerConfiguration(implicit conn: Connection) =

ledger/sandbox/src/main/scala/com/digitalasset/platform/store/dao/LedgerDao.scala

@@ -160,7 +160,7 @@ trait LedgerWriteDao extends ReportsHealth {
     *
     * @param ledgerId the ledger id to be stored
     */
-  def initializeLedger(ledgerId: LedgerId, ledgerEnd: Offset): Future[Unit]
+  def initializeLedger(ledgerId: LedgerId): Future[Unit]
 
   def storeTransaction(
       submitterInfo: Option[SubmitterInfo],

ledger/sandbox/src/main/scala/com/digitalasset/platform/store/dao/MeteredLedgerDao.scala

@@ -168,8 +168,8 @@ class MeteredLedgerDao(ledgerDao: LedgerDao, metrics: Metrics)
       metrics.daml.index.db.storeInitialState,
       ledgerDao.storeInitialState(ledgerEntries, newLedgerEnd))
 
-  override def initializeLedger(ledgerId: LedgerId, ledgerEnd: Offset): Future[Unit] =
-    ledgerDao.initializeLedger(ledgerId, ledgerEnd)
+  override def initializeLedger(ledgerId: LedgerId): Future[Unit] =
+    ledgerDao.initializeLedger(ledgerId)
 
   override def reset(): Future[Unit] =
     ledgerDao.reset()

ledger/sandbox/src/test/lib/scala/com/digitalasset/platform/store/dao/JdbcLedgerDaoBackend.scala

@@ -5,7 +5,6 @@ package com.daml.platform.store.dao
 
 import com.codahale.metrics.MetricRegistry
 import com.daml.caching.Cache
-import com.daml.ledger.participant.state.v1.Offset
 import com.daml.ledger.api.domain.LedgerId
 import com.daml.ledger.api.testing.utils.AkkaBeforeAndAfterAll
 import com.daml.logging.LoggingContext
@@ -46,7 +45,7 @@ private[dao] trait JdbcLedgerDaoBackend extends AkkaBeforeAndAfterAll { this: Su
       for {
         _ <- Resource.fromFuture(new FlywayMigrations(jdbcUrl).migrate())
         dao <- daoOwner.acquire()
-        _ <- Resource.fromFuture(dao.initializeLedger(LedgerId("test-ledger"), Offset.begin))
+        _ <- Resource.fromFuture(dao.initializeLedger(LedgerId("test-ledger")))
       } yield dao
     }
     ledgerDao = Await.result(resource.asFuture, 10.seconds)