Utility for migrating from memiavl to flatKV#3275
Merged
cody-littley merged 27 commits intomainfrom Apr 23, 2026
Merged
Conversation
added 17 commits
April 17, 2026 08:14
|
The latest Buf updates on your PR. Results from workflow Buf / buf (pull_request).
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #3275 +/- ##
==========================================
+ Coverage 59.11% 59.30% +0.18%
==========================================
Files 2085 2076 -9
Lines 170874 169824 -1050
==========================================
- Hits 101019 100706 -313
+ Misses 61059 60363 -696
+ Partials 8796 8755 -41
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
yzang2019
approved these changes
Apr 22, 2026
| newErr = e | ||
| newDone = true | ||
| case <-ctx.Done(): | ||
| return ctx.Err() |
Contributor
There was a problem hiding this comment.
Do we need to drain both channels before returning here?
Contributor
Author
There was a problem hiding this comment.
If the context gets cancelled, this is basically a rugpull, and my goal is to tear down everything to avoid deadlocked threads. IMO, we should treat an unexpected context cancel equivalent to a "we're crashing now, free up everything in preparation for that".
The most useful time for this sort of teardown behavior is in unit tests. If we always use t.Context(), then we ensure that we don't leave behind deadlocked threads when a test ends. The goal is to make it so that each thread has a fast teardown mechanism for when we want to tear it down.
| case MigrationComplete: | ||
| return len(treeNames) | ||
| } | ||
| return sort.Search(len(treeNames), func(i int) bool { |
Contributor
There was a problem hiding this comment.
I don't think tree name is guaranteed to be sorted?
Contributor
Author
There was a problem hiding this comment.
They are currently sorted. But you make a good point, it's fragile for us to assume they will always be sorted. I added a defensive sort just in case.
| // Write the value to the new DB. | ||
| putPair(newDBPairsByStore, value.ModuleName, &proto.KVPair{Key: value.Key, Value: value.Value}) | ||
| // Delete the value from the old DB. | ||
| putPair(oldDBPairsByStore, value.ModuleName, &proto.KVPair{Key: value.Key, Delete: true}) |
Contributor
There was a problem hiding this comment.
Would recommend add a metrics how how many keys are migrated successfully
Contributor
Author
There was a problem hiding this comment.
Metrics added. The count metric will get reset after a restart or a state sync, but that should be ok as long as we are aware that it works that way.
blindchaser
reviewed
Apr 23, 2026
Contributor
There was a problem hiding this comment.
do we want to commit the binaries here?
Contributor
Author
There was a problem hiding this comment.
No, we don't! Removed.
This happened when I switched branches. It's was in the .gitignore in that branch, but not this one.
blindchaser
reviewed
Apr 23, 2026
blindchaser
reviewed
Apr 23, 2026
| // positioned at the start (boundary defaults to MigrationBoundaryNotStarted). | ||
| // If autoRebuild is true, the iterator re-reads from Data before every | ||
| // NextBatch call, so external mutations are picked up automatically. | ||
| func NewMapMigrationIterator(data map[string]map[string][]byte, autoRebuild bool) *MapMigrationIterator { |
Contributor
There was a problem hiding this comment.
looks this only being used in the testing files? how about rename it to mock_migration_iterator_test.go
blindchaser
reviewed
Apr 23, 2026
| if m == nil || m.boundarySnapshot == nil { | ||
| return | ||
| } | ||
| go func() { |
Contributor
There was a problem hiding this comment.
the goroutine is fire-and-forget. do we want to add some done chan struct{} or sync.WaitGroup to expose()a Close() method or something?
Contributor
Author
There was a problem hiding this comment.
The goroutine dies when the context is cancelled. But it's not tricky to add an explicit Close() method. Added.
blindchaser
reviewed
Apr 23, 2026
Comment on lines
+1
to
+454
| package migration | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/binary" | ||
| "errors" | ||
| "fmt" | ||
| "sort" | ||
| "time" | ||
|
|
||
| "github.com/sei-protocol/sei-chain/sei-db/proto" | ||
| "github.com/sei-protocol/seilog" | ||
| ) | ||
|
|
||
| var logger = seilog.NewLogger("db", "state-db", "sc", "migration") | ||
|
|
||
| // MigrationManager handles migration from one database to another, | ||
| // routing reads and writes during the course of the migration. | ||
| // | ||
| // MigrationManager supports concurrent Read calls. ApplyChangeSets must not | ||
| // run concurrently with Read or with itself. | ||
| // | ||
| // A migration manager has two states: "migrating" and "passthrough". | ||
| // | ||
| // 1. Migrating: active migration. Reads split across old/new DBs by | ||
| // boundary, writes are routed across the boundary and applied to | ||
| // both DBs in parallel. Each block, N keys are deleted from the old DB | ||
| // and written to the new DB. | ||
| // 2. Passthrough: All reads/writes forwarded directly | ||
| // to the new DB. No boundary, no iterator. | ||
| type MigrationManager struct { | ||
|
|
||
| // For reading values out of the old database. May be nil once the | ||
| // manager is in the passthrough state (post-finalization or | ||
| // constructed at targetVersion). | ||
| oldDBReader DBReader | ||
|
|
||
| // For writing values to the old database. May be nil in passthrough | ||
| // (see oldDBReader). | ||
| oldDBWriter DBWriter | ||
|
|
||
| // For reading values out of the new database. | ||
| newDBReader DBReader | ||
|
|
||
| // For writing values to the new database. | ||
| newDBWriter DBWriter | ||
|
|
||
| // For iterating through key-value pairs to migrate in the old | ||
| // database. May be nil in passthrough. | ||
| iterator MigrationIterator | ||
|
|
||
| // The boundary of the migration. All keys to the left of (or equal | ||
| // to) the boundary are considered migrated. In passthrough this is | ||
| // pinned to MigrationBoundaryComplete, though Read short-circuits | ||
| // via migrationFinished before consulting the boundary. | ||
| boundary MigrationBoundary | ||
|
|
||
| // The number of key-value pairs to migrate after each write operation. | ||
| migrationBatchSize int | ||
|
|
||
| // The version we want to migrate to. | ||
| targetVersion uint64 | ||
|
|
||
| // If true, then the migration has been fully completed. | ||
| migrationFinished bool | ||
|
|
||
| // Optional metrics sink. May be nil; all calls on this field go | ||
| // through nil-safe methods on *MigrationMetrics. | ||
| metrics *MigrationMetrics | ||
| } | ||
|
|
||
| // Handles the migration of data from one database to another. | ||
| func NewMigrationManager( | ||
| // The number of key-value pairs to migrate after each write operation. Must be > 0. | ||
| migrationBatchSize int, | ||
| // The migration version the stored data is expected to be at on entry. If no prior migration | ||
| // version is stored in the DB, startVersion should be 0. | ||
| startVersion uint64, | ||
| // The migration version after the migration is complete. | ||
| // Must be strictly greater than startVersion. | ||
| targetVersion uint64, | ||
| // For reading values out of the old database. May be nil iff the new | ||
| // DB already reports targetVersion. | ||
| oldDBReader DBReader, | ||
| // For writing values to the old database. May be nil iff the new DB | ||
| // already reports targetVersion. | ||
| oldDBWriter DBWriter, | ||
| // For reading values out of the new database. | ||
| newDBReader DBReader, | ||
| // For writing values to the new database. | ||
| newDBWriter DBWriter, | ||
| // For iterating through key-value pairs to migrate in the old | ||
| // database. May be nil iff the new DB already reports targetVersion. | ||
| iterator MigrationIterator, | ||
| // Optional metrics sink. Pass nil to disable metric emission. | ||
| metrics *MigrationMetrics, | ||
| ) (*MigrationManager, error) { | ||
|
|
||
| // Always-required handles and parameters. | ||
| if newDBReader == nil { | ||
| return nil, errors.New("newDBReader must not be nil") | ||
| } | ||
| if newDBWriter == nil { | ||
| return nil, errors.New("newDBWriter must not be nil") | ||
| } | ||
| if migrationBatchSize <= 0 { | ||
| return nil, fmt.Errorf("migration batch size must be positive, got %d", migrationBatchSize) | ||
| } | ||
| if startVersion >= targetVersion { | ||
| return nil, fmt.Errorf("startVersion (%d) must be strictly less than targetVersion (%d)", | ||
| startVersion, targetVersion) | ||
| } | ||
|
|
||
| // Look up the version from the new DB first. If it's already at | ||
| // targetVersion the migration has completed on a prior boot and we | ||
| // don't need the old DB for anything. | ||
| currentMigrationVersion, versionKnown, err := readVersionFromDB(newDBReader) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to read migration version from new DB: %w", err) | ||
| } | ||
|
|
||
| if versionKnown { | ||
| if currentMigrationVersion == targetVersion { | ||
| // Passthrough path, migration already complete. | ||
| logger.Info("migration manager constructed in passthrough mode", "targetVersion", targetVersion) | ||
| metrics.SetVersion(targetVersion) | ||
| metrics.SetBoundary(MigrationBoundaryComplete) | ||
| return &MigrationManager{ | ||
| newDBReader: newDBReader, | ||
| newDBWriter: newDBWriter, | ||
| boundary: MigrationBoundaryComplete, | ||
| migrationBatchSize: migrationBatchSize, | ||
| targetVersion: targetVersion, | ||
| migrationFinished: true, | ||
| metrics: metrics, | ||
| }, nil | ||
| } | ||
| if currentMigrationVersion != startVersion { | ||
| return nil, fmt.Errorf( | ||
| "unexpected migration version in new DB: expected %d (start) or %d (target), got %d", | ||
| startVersion, targetVersion, currentMigrationVersion) | ||
| } | ||
| } | ||
|
|
||
| // Migration is not complete, so we can't tolerate nil old DB accessors. | ||
| if oldDBReader == nil { | ||
| return nil, errors.New("oldDBReader must not be nil when new DB is not at targetVersion") | ||
| } | ||
| if oldDBWriter == nil { | ||
| return nil, errors.New("oldDBWriter must not be nil when new DB is not at targetVersion") | ||
| } | ||
| if iterator == nil { | ||
| return nil, errors.New("iterator must not be nil when new DB is not at targetVersion") | ||
| } | ||
|
|
||
| if !versionKnown { | ||
| // The version wasn't in the new DB, so read it from the old DB. | ||
| currentMigrationVersion, _, err = readVersionFromDB(oldDBReader) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to read migration version from old DB: %w", err) | ||
| } | ||
| if currentMigrationVersion != startVersion { | ||
| return nil, fmt.Errorf( | ||
| "unexpected migration version in old DB: expected %d, got %d", startVersion, currentMigrationVersion) | ||
| } | ||
| } | ||
|
|
||
| boundary, err := readMigrationBoundary(newDBReader) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to read migration boundary: %w", err) | ||
| } | ||
| iterator.SetBoundary(boundary) | ||
|
|
||
| logger.Info("initialized migration manager", | ||
| "startVersion", startVersion, | ||
| "targetVersion", targetVersion, | ||
| "boundary", boundary.String()) | ||
|
|
||
| metrics.SetVersion(currentMigrationVersion) | ||
| metrics.SetBoundary(boundary) | ||
|
|
||
| return &MigrationManager{ | ||
| oldDBReader: oldDBReader, | ||
| oldDBWriter: oldDBWriter, | ||
| newDBReader: newDBReader, | ||
| newDBWriter: newDBWriter, | ||
| iterator: iterator, | ||
| boundary: boundary, | ||
| migrationBatchSize: migrationBatchSize, | ||
| targetVersion: targetVersion, | ||
| metrics: metrics, | ||
| }, nil | ||
| } | ||
|
|
||
| // readMigrationBoundary reads the current migration boundary from the new | ||
| // database, or returns MigrationBoundaryNotStarted if none is stored yet. | ||
| func readMigrationBoundary(newDBReader DBReader) (MigrationBoundary, error) { | ||
| boundaryBytes, ok, err := newDBReader(MigrationStore, []byte(MigrationBoundaryKey)) | ||
| if err != nil { | ||
| return MigrationBoundary{}, fmt.Errorf("failed to get migration boundary: %w", err) | ||
| } | ||
| if !ok { | ||
| return MigrationBoundaryNotStarted, nil | ||
| } | ||
| boundary, err := DeserializeMigrationBoundary(boundaryBytes) | ||
| if err != nil { | ||
| return MigrationBoundary{}, fmt.Errorf("failed to deserialize migration boundary: %w", err) | ||
| } | ||
| return boundary, nil | ||
| } | ||
|
|
||
| // readVersionFromDB reads MigrationVersionKey from the given DB's | ||
| // MigrationStore, returning (version, present, error). An absent key is | ||
| // reported as (0, false, nil) so the caller can distinguish "not set" | ||
| // from "explicitly zero". | ||
| // | ||
| // This helper takes a raw DBReader rather than going through | ||
| // MigrationManager.Read because the MigrationStore is reserved for | ||
| // internal use and MigrationManager.Read rejects reads against it. | ||
| func readVersionFromDB(reader DBReader) (uint64, bool, error) { | ||
| data, ok, err := reader(MigrationStore, []byte(MigrationVersionKey)) | ||
| if err != nil { | ||
| return 0, false, err | ||
| } | ||
| if !ok { | ||
| return 0, false, nil | ||
| } | ||
| if len(data) != 8 { | ||
| return 0, false, fmt.Errorf( | ||
| "expected 8-byte migration version, got %d bytes", len(data)) | ||
| } | ||
| return binary.BigEndian.Uint64(data), true, nil | ||
| } | ||
|
|
||
| // IsAtVersion reports whether the DB reached by reader is currently at the | ||
| // given migration version. An absent MigrationVersionKey is interpreted as | ||
| // version 0. | ||
| // | ||
| // Intended for callers that need to decide, before constructing a | ||
| // MigrationManager, whether to bother opening the legacy/old DB at all: | ||
| // | ||
| // atTarget, err := migration.IsAtVersion(newReader, targetVersion) | ||
| // if err != nil { /* handle */ } | ||
| // if atTarget { | ||
| // // Skip opening the old DB; just go straight to the new one. | ||
| // } | ||
| // | ||
| // This is a pure lookup; it does not mutate state or call any finalizer. | ||
| func IsAtVersion(reader DBReader, version uint64) (bool, error) { | ||
| v, _, err := readVersionFromDB(reader) | ||
| if err != nil { | ||
| return false, err | ||
| } | ||
| return v == version, nil | ||
| } | ||
|
|
||
| // Read a value from the database. If the requested value is migrated, read it from the new database. | ||
| // Otherwise, read it from the old database. | ||
| // | ||
| // Reads targeting MigrationStore are rejected with an error: that store | ||
| // is reserved for the manager's own bookkeeping. | ||
| // | ||
| // In passthrough (migrationFinished=true), all reads route to the new DB. | ||
| // | ||
| // Not safe to call concurrently with ApplyChangeSets. | ||
| func (m *MigrationManager) Read(store string, key []byte) ([]byte, bool, error) { | ||
| if store == MigrationStore { | ||
| // The migration module is reserved for internal use, do not permit outer scope reads from it. | ||
| return nil, false, fmt.Errorf("reads from the 'migration' module are not permitted") | ||
| } | ||
| if m.migrationFinished { | ||
| // We've finished the migration, all reads should go to the new DB. | ||
| return m.newDBReader(store, key) | ||
| } | ||
| if m.boundary.IsMigrated(store, key) { | ||
| // We are mid-migration and this key has already been migrated, read it from the new DB. | ||
| return m.newDBReader(store, key) | ||
| } | ||
| // We are mid-migration and this key has not been migrated, read it from the old DB. | ||
| return m.oldDBReader(store, key) | ||
| } | ||
|
|
||
| // ApplyChangeSets applies a batch of change sets to the database. | ||
| // | ||
| // Not safe to call concurrently with Read or itself. | ||
| func (m *MigrationManager) ApplyChangeSets(ctx context.Context, changesets []*proto.NamedChangeSet) error { | ||
| start := time.Now() | ||
| defer func() { | ||
| m.metrics.RecordApplyDuration(time.Since(start)) | ||
| }() | ||
|
|
||
| if changesets == nil { | ||
| changesets = make([]*proto.NamedChangeSet, 0) | ||
| } | ||
| for _, cs := range changesets { | ||
| if cs.Name == MigrationStore { | ||
| // The migration module is reserved for internal use, do not permit outer scope writes to it. | ||
| return fmt.Errorf("writes to internal migration store %q are not permitted", MigrationStore) | ||
| } | ||
| } | ||
|
|
||
| if m.migrationFinished { | ||
| // Passthrough: migration is complete. | ||
| if err := m.newDBWriter(changesets); err != nil { | ||
| return fmt.Errorf("failed to apply changes to new database: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Get the next batch of keys to migrate. | ||
| valuesToMigrate, newBoundary, err := m.iterator.NextBatch(m.migrationBatchSize) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to get next batch: %w", err) | ||
| } | ||
| m.boundary = newBoundary | ||
| m.migrationFinished = newBoundary.Status() == MigrationComplete | ||
| m.metrics.SetBoundary(newBoundary) | ||
|
|
||
| // Pairs destined for each DB, grouped by store name and keyed by KVPair.Key. | ||
| oldDBPairsByStore := make(map[string]map[string]*proto.KVPair) | ||
| newDBPairsByStore := make(map[string]map[string]*proto.KVPair) | ||
|
|
||
| // Create change sets that move the values to migrate from the old DB to the new DB. | ||
| var keyBytesThisBatch, valueBytesThisBatch int64 | ||
| for _, value := range valuesToMigrate { | ||
| keyBytesThisBatch += int64(len(value.Key)) | ||
| valueBytesThisBatch += int64(len(value.Value)) | ||
| // Write the value to the new DB. | ||
| putPair(newDBPairsByStore, value.ModuleName, &proto.KVPair{Key: value.Key, Value: value.Value}) | ||
| // Delete the value from the old DB. | ||
| putPair(oldDBPairsByStore, value.ModuleName, &proto.KVPair{Key: value.Key, Delete: true}) | ||
| } | ||
| m.metrics.ReportKeysMigrated(int64(len(valuesToMigrate)), keyBytesThisBatch, valueBytesThisBatch) | ||
|
|
||
| // For each pair in the original change sets, route to the appropriate database. | ||
| // These must overwrite migrated values, so it's important to do this after we've collected | ||
| // the change set for the migrated values. | ||
| for _, changeSet := range changesets { | ||
| for _, pair := range changeSet.Changeset.Pairs { | ||
| if m.boundary.IsMigrated(changeSet.Name, pair.Key) { | ||
| putPair(newDBPairsByStore, changeSet.Name, pair) | ||
| } else { | ||
| putPair(oldDBPairsByStore, changeSet.Name, pair) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| oldDBChangeSet := flattenPairsByStore(oldDBPairsByStore) | ||
| newDBChangeSets := flattenPairsByStore(newDBPairsByStore) | ||
|
|
||
| if m.migrationFinished { | ||
| // On the final block of the migration, update the migration version and delete the boundary. | ||
| versionBytes := make([]byte, 8) | ||
| binary.BigEndian.PutUint64(versionBytes, m.targetVersion) | ||
| newDBChangeSets = append(newDBChangeSets, &proto.NamedChangeSet{ | ||
| Name: MigrationStore, | ||
| Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{ | ||
| {Key: []byte(MigrationVersionKey), Value: versionBytes}, | ||
| {Key: []byte(MigrationBoundaryKey), Delete: true}, | ||
| }}, | ||
| }) | ||
| // Mirror the on-disk version bump in the in-memory metric so the | ||
| // version gauge and the boundary-snapshot loop see the | ||
| // completion at the same moment the DB does. | ||
| m.metrics.SetVersion(m.targetVersion) | ||
| } else { | ||
| // On every other block of the migration, update the boundary. | ||
| newDBChangeSets = append(newDBChangeSets, &proto.NamedChangeSet{ | ||
| Name: MigrationStore, | ||
| Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{ | ||
| {Key: []byte(MigrationBoundaryKey), Value: newBoundary.Serialize()}, | ||
| }}, | ||
| }) | ||
| } | ||
|
|
||
| // Apply changes to each database in parallel. | ||
| oldDBErr := make(chan error, 1) | ||
| newDBErr := make(chan error, 1) | ||
| go func() { | ||
| err := m.oldDBWriter(oldDBChangeSet) | ||
| if err != nil { | ||
| err = fmt.Errorf("failed to apply changes to old database: %w", err) | ||
| } | ||
| oldDBErr <- err | ||
| }() | ||
| go func() { | ||
| err := m.newDBWriter(newDBChangeSets) | ||
| if err != nil { | ||
| err = fmt.Errorf("failed to apply changes to new database: %w", err) | ||
| } | ||
| newDBErr <- err | ||
| }() | ||
|
|
||
| // Wait for both writers to finish. | ||
| var oldErr, newErr error | ||
| oldDone, newDone := false, false | ||
| for !oldDone || !newDone { | ||
| select { | ||
| case e := <-oldDBErr: | ||
| oldErr = e | ||
| oldDone = true | ||
| case e := <-newDBErr: | ||
| newErr = e | ||
| newDone = true | ||
| case <-ctx.Done(): | ||
| return ctx.Err() | ||
| } | ||
| } | ||
|
|
||
| if err := errors.Join(oldErr, newErr); err != nil { | ||
| return fmt.Errorf("failed to apply changes to databases: %w", err) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // putPair inserts pair into dest under (store, pair.Key), creating the inner | ||
| // map on demand. Later writes to the same (store, key) overwrite earlier ones. | ||
| func putPair(dest map[string]map[string]*proto.KVPair, store string, pair *proto.KVPair) { | ||
| byKey, ok := dest[store] | ||
| if !ok { | ||
| byKey = make(map[string]*proto.KVPair) | ||
| dest[store] = byKey | ||
| } | ||
| byKey[string(pair.Key)] = pair | ||
| } | ||
|
|
||
| // flattenPairsByStore collapses a store-keyed map of (key -> KVPair) into one | ||
| // NamedChangeSet per store, with stores and pairs emitted in sorted order for | ||
| // deterministic downstream writes. | ||
| func flattenPairsByStore(pairsByStore map[string]map[string]*proto.KVPair) []*proto.NamedChangeSet { | ||
| storeNames := make([]string, 0, len(pairsByStore)) | ||
| for name := range pairsByStore { | ||
| storeNames = append(storeNames, name) | ||
| } | ||
| sort.Strings(storeNames) | ||
|
|
||
| changeSets := make([]*proto.NamedChangeSet, 0, len(storeNames)) | ||
| for _, name := range storeNames { | ||
| byKey := pairsByStore[name] | ||
| pairs := make([]*proto.KVPair, 0, len(byKey)) | ||
| for _, pair := range byKey { | ||
| pairs = append(pairs, pair) | ||
| } | ||
| sort.Slice(pairs, func(i, j int) bool { | ||
| return string(pairs[i].Key) < string(pairs[j].Key) | ||
| }) | ||
| changeSets = append(changeSets, &proto.NamedChangeSet{ | ||
| Name: name, | ||
| Changeset: proto.ChangeSet{Pairs: pairs}, | ||
| }) | ||
| } | ||
| return changeSets | ||
| } |
Contributor
There was a problem hiding this comment.
during block processing, could queries arrive while ApplyChangeSets runs? do we need a RWMutex here
Contributor
Author
There was a problem hiding this comment.
Requires discussion.
Although it is currently thread safe to read values concurrently with a write, we don't have API for atomic batch reads. Even without this code, a batch read might return an inconsistant slice of data (i.e. some data from block N, other data from block N+1).
Contributor
Author
There was a problem hiding this comment.
Added mutex to make this safe.
blindchaser
reviewed
Apr 23, 2026
| buf := make([]byte, 1+4+len(nameBytes)+len(mb.key)) | ||
| buf[0] = byte(mb.status) | ||
| binary.BigEndian.PutUint32(buf[1:5], uint32(len(nameBytes))) //nolint:gosec // module names are short strings | ||
| copy(buf[5:5+len(nameBytes)], nameBytes) |
Contributor
There was a problem hiding this comment.
the number 5 (1-byte status + 4-byte length prefix) appears ~8 times across Serialize and DeserializeMigrationBoundary. consider add single constant (like const headerSize = 5)
blindchaser
approved these changes
Apr 23, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe your changes and provide context
This PR introduces a utility for migrating data from database A to database B. It's generic so that we can encapsulate migration logic. Additionally, we can reuse this utility in the future if we ever have a future migration of similar nature.
Testing performed to validate your change
Lots of unit tests.