[ENH] Add log sealing to the go service. (#4554)

## Description of changes

This PR does the following:
- Adds a migration to add the `is_sealed` column to the collection table
  in the log database.
- Updates the go/ README.md to mention the correct atlas command to run.
- Updates the go store queries to recognize the new field.
- Add a new SealLog query to unconditionally seal a log.
- Propagate the `is_sealed` column from table to `IsSealed` field in the
  protobuf message.
- Add a test that push logs will fail when sealed.
- Add support to the RustLogService for the new IDL.
- Translate `is_sealed` responses into explicit errors in the log
  client.

## Test plan

CI, as usual.

## Documentation Changes

N/A

Author: rescrv

go/README.md

@@ -35,3 +35,12 @@ ALSO, ensure you have copied the `/include` directory of the release to `../incl
 Then, to install the plugins, run the `go install` commands from the `Dockerfile`. The exact commands are not here because we would be duplicating where versions live if we did. The `Dockerfile` is the source of truth for the versions.
 
 Once those are all installed, you can run `make build` to build the project and most importantly, the generated protobuf files which your IDE will complain about until they are generated.
+
+## Schema Migrations
+
+From the directory with the migrations/ and schema/ directories, you can generate a new schema by
+changing the files in schema directly and running this command:
+
+```
+atlas migrate diff --dir file://migrations --to file://schema --dev-url 'docker://postgres/15/dev?search_path=public'
+```

go/pkg/log/repository/log.go

@@ -23,7 +23,7 @@ type LogRepository struct {
 	sysDb   sysdb.ISysDB
 }
 
-func (r *LogRepository) InsertRecords(ctx context.Context, collectionId string, records [][]byte) (insertCount int64, err error) {
+func (r *LogRepository) InsertRecords(ctx context.Context, collectionId string, records [][]byte) (insertCount int64, isSealed bool, err error) {
 	var tx pgx.Tx
 	tx, err = r.conn.BeginTx(ctx, pgx.TxOptions{})
 	if err != nil {
@@ -66,6 +66,13 @@ func (r *LogRepository) InsertRecords(ctx context.Context, collectionId string,
 			return
 		}
 	}
+	if collection.IsSealed {
+		insertCount = 0
+		isSealed = true
+		err = nil
+		return
+	}
+	isSealed = false
 	params := make([]log.InsertRecordParams, len(records))
 	for i, record := range records {
 		offset := collection.RecordEnumerationOffsetPosition + int64(i) + 1

go/pkg/log/repository/log_test.go

@@ -45,13 +45,15 @@ func (suite *LogTestSuite) TestGarbageCollection() {
 	collectionID2 := types.NewUniqueID()
 
 	// Add records to collection 1
-	count, err := suite.lr.InsertRecords(ctx, collectionID1.String(), [][]byte{{1, 2, 3}})
+	count, isSealed, err := suite.lr.InsertRecords(ctx, collectionID1.String(), [][]byte{{1, 2, 3}})
 	assert.NoError(suite.t, err, "Failed to insert records")
+	assert.False(suite.t, isSealed, count, "Log sealed")
 	assert.Equal(suite.t, int64(1), count, "Failed to insert records")
 
 	// Add records to collection 2
-	count, err = suite.lr.InsertRecords(ctx, collectionID2.String(), [][]byte{{1, 2, 3}})
+	count, isSealed, err = suite.lr.InsertRecords(ctx, collectionID2.String(), [][]byte{{1, 2, 3}})
 	assert.NoError(suite.t, err, "Failed to insert records")
+	assert.False(suite.t, isSealed, count, "Log sealed")
 	assert.Equal(suite.t, int64(1), count, "Failed to insert records")
 
 	// Add collection 1 to sysdb
@@ -72,8 +74,9 @@ func (suite *LogTestSuite) TestGarbageCollection() {
 	assert.Equal(suite.t, 0, len(records), "Failed to run garbage collection")
 
 	// Add records to collection 2, expect offset to reset
-	count, err = suite.lr.InsertRecords(ctx, collectionID2.String(), [][]byte{{4, 5, 6}})
+	count, isSealed, err = suite.lr.InsertRecords(ctx, collectionID2.String(), [][]byte{{4, 5, 6}})
 	assert.NoError(suite.t, err, "Failed to insert records")
+	assert.False(suite.t, isSealed, count, "Log sealed")
 	assert.Equal(suite.t, int64(1), count, "Failed to insert records")
 	records, err = suite.lr.PullRecords(ctx, collectionID2.String(), 1, 1, time.Now().UnixNano())
 	assert.NoError(suite.t, err, "Failed to pull records")
@@ -110,6 +113,24 @@ func (suite *LogTestSuite) TestUniqueConstraintPushLogs() {
 	}
 }
 
+func (suite *LogTestSuite) TestSealedLogWontPush() {
+	ctx := context.Background()
+	collectionId := types.NewUniqueID()
+	params := log.InsertCollectionParams {
+		ID: collectionId.String(),
+		RecordEnumerationOffsetPosition: 1,
+		RecordCompactionOffsetPosition: 0,
+	}
+	_, err := suite.lr.queries.InsertCollection(ctx, params)
+	assert.NoError(suite.t, err, "Initializing log should not fail.")
+	_, err = suite.lr.queries.SealLog(ctx, collectionId.String())
+	assert.NoError(suite.t, err, "Sealing log should not fail.")
+	var isSealed bool
+	_, isSealed, err = suite.lr.InsertRecords(ctx, collectionId.String(), [][]byte{{1,2,3}})
+	assert.NoError(suite.t, err, "Failed to push logs")
+	assert.True(suite.t, isSealed, "Did not report was sealed")
+}
+
 func TestLogTestSuite(t *testing.T) {
 	testSuite := new(LogTestSuite)
 	testSuite.t = t

go/pkg/log/server/server.go

@@ -36,12 +36,14 @@ func (s *logServer) PushLogs(ctx context.Context, req *logservicepb.PushLogsRequ
 		recordsContent = append(recordsContent, data)
 	}
 	var recordCount int64
-	recordCount, err = s.lr.InsertRecords(ctx, collectionID.String(), recordsContent)
+	var isSealed bool
+	recordCount, isSealed, err = s.lr.InsertRecords(ctx, collectionID.String(), recordsContent)
 	if err != nil {
 		return
 	}
 	res = &logservicepb.PushLogsResponse{
 		RecordCount: int32(recordCount),
+		LogIsSealed: isSealed,
 	}
 	return
 }

go/pkg/log/store/db/copyfrom.go

@@ -0,0 +1,2 @@
+-- Modify "collection" table
+ALTER TABLE "collection" ADD COLUMN "is_sealed" boolean NOT NULL DEFAULT false;

go/pkg/log/store/db/db.go

@@ -1,2 +1,3 @@
-h1:kG+ejV1DS3youx+m5SNNFYabJeDqfYTdSQHbJtR2/eU=
+h1:Ha4eNnpWJQC0flnfRZKnpw06LblDCh5ocPykLIy5koE=
 20240404181827_initial.sql h1:xnoD1FcXImqQPJOvaDbTOwTGPLtCP3RibetuaaZeATI=
+20250515150704.sql h1:ySONZilpdnd0BHQhsVAoSj5ud7/3gpI03VuwwWqZqrE=