By Oliver Nguyen
As a start-up company, we started with a simple architecture: All data were stored in a single centralized Postgres instance, shared by a few services. It has been working well and allows us to move fast, deliver features, enter the market, acquire clients, and grow exponentially.
Fast forward a couple of years, today we have thousands of clients and billions of records, it's time to migrate the top tables from Postgres to a better solution. We choose DynamoDB, a key-value datastore by AWS, with high availability, low-latency performance, and see the queries reduce to single-digit milliseconds.
As with any decent migration plan, we carefully analyze usages, define schema and indexes, start with double-write, switch read queries to DynamoDB with fallback to Postgres, import all records to DynamoDB, and finally stop querying from Postgres.
This article focuses on the step of importing records for the room_events table: Scan all the records of this table and write them to DynamoDB*. The goal is to ensure a complete migration, with no events missed. The migration script must be fast, capable of running in parallel, and able to stop and resume from its last state. It should also be resilient, handling network errors or interruptions and resuming seamlessly.*
Let's see how we can do this! 🥸
Here is the simplified schema of the room_events table in Postgres:
room_events
Notes:
Let's write a simple version of the migration script: Run in a single loop, scan all records, and write to DynamoDB. Also, include a few trivial things to not have to worry about them later:
import "connectly.ai/go/pkgs/ulid"
var startDate = parseTime("2020-01-01T00:00:00Z")
var endDate = parseTime("2024-10-20T00:00:00Z")
var pgBatchSize = 1000
var dynBatchSize = 25
type Migrator struct {
*logger
pg PostgresClient
dyn DynamoClient
}
type QueryRoomEventsRequest struct {
BeforeID ulid.ULID
Limit int
}
type QueryRoomEventsResponse struct {
Items []RoomEvents
lastID ulid.ULID
}
func (m *Migrator) Migrate(ctx context.Context) error {
lastID, count := ulid.FromTime(endDate), 0
for {
req := QueryRoomEventsRequest{
BeforeID: lastID,
Limit: pgBatchSize
}
res, err := m.queryRoomEvents(ctx, req)
m.Logger(ctx).Must(err, "failed to query room events")
count += res.Items
if len(res.Items) == 0 {
m.Logger(ctx).Infof("MIGRATION DONE: count=%v", count)
return nil
}
for i := 0; i < len(res.Items); i += dynBatchSize {
j := min(i + dynBatchSize, len(res.Items))
batch := res.Items[i, j]
_, err = m.batchWriteDynamo(ctx, batch)
m.Logger(ctx).Must(err, "failed to write to dynamodb")
}
}
}
func (m *Migrator) queryRoomEvents( /* ... */ ) { /* ... */ }
func (m *Migrator) batchWriteDynamo(/* ... */ ) { /* ... */ }
This naive script will operate under some assumptions:
However, in the real world, we can not rely on those assumptions:
To handle these challenges, we need to improve our script and implement a more robust, distributed solution:
Data Splitting and Partitioning:
Parallel Processing with Multiple Workers:
States Management in Redis:
Keeper for Centralized Progress Tracking:
Timeout Handling and Retry Mechanism:
Data Verification:
Leveraging Idempotency in Writes:
Another important point is that the write step for each record is idempotent. This means that writing the same record to DynamoDB multiple times will simply overwrite the previous version, ensuring no duplicates.
We can use this to our advantage:
With those concepts in mind, let's start thinking about the architecture.
Given that we already split the records into Timeslots, each consists of all room events for one hour. We need to keep track of the progress of these slots in Redis:
So we need 2 keys for each slot. Multiply with 3 years of data, there will be 52.560 keys (3 * 365 * 24 * 2). That's a lot!
We can optimize the Redis keys by cleaning the group of consecutive FINISHED slots and replacing them with a single mgre:last_slot key. This way, we only need to keep track of a small amount of keys. Of course, this is based on the assumption that the slots have a similar number of room events, so each worker can take a similar amount of time to finish.
The script runs in multiple pods. Each pod consists of a Manager, a Keeper , and multiple Workers. Each run in its own goroutine.
A Manager manages all workers in a pod:
A Worker migrate room events by each Timeslot:
A Keeper overview the whole migration progress:
func (m *Migrator) runManager(ctx context.Context) {
// ... recover, logs, metrics ...
// ... loop and call runManagerStep (more on this later) ...
}
func (m *Migrator) runManagerStep(ctx context.Context) Status {
// ... recover, logs, metrics ...
for {
mustRefresh(ctx, m, m.globalStatus, m.lastSlot)
if m.globalStatus.Load().Is(SKIPPED, FINISHED) { return /*...*/ }
for slot := m.lastSlot.Load(); slot.Before(endSlot); slot.Next() {
slotStates := m.getSlotStates(slot)
mustRefresh(ctx, m, slotStates)
if slotStates.Status.Load().Is(SKIPPED, FINISHED) { continue }
slotWorkerID := mustAcquireLock(ctx, m, slot.Worker)
if !slotWorkerID.Load().IsPod(m.PodID) { continue }
select {
case <- ctx.Done():
return IN_PROGRESS // 👈 stop when the pod stops
case m.slotCh <- slot:
continue // 👈 send to slot channel
}
}
}
// 👇 still IN_PROGRESS, only Keeper can verify all slots are FINISHED
return IN_PROGRESS
}
func (m *Migrator) runWorker(ctx context.Context) {
// ... recover, logs, metrics ...
for {
select {
case <- ctx.Done():
return
case slot := <- m.slotCh
s.runTimeslot(ctx, slot)
}
}
}
func (m *Migrator) runTimeslot(ctx context.Context, slot Timeslot) {
// ... recover, logs, metrics ...
t := time.NewTimer(0)
for {
select {
case <- ctx.Done():
return
case <- t.C
status := s.runTimeslotStep(ctx, slot)
if status.Is(SKIPPED, FINISHED) { return }
t.Reset(3 * time.Second)
}
}
}
func (m *Migrator) runTimeslotStep(ctx context.Context, slot Timeslot) Status {
// ... recover, logs, metrics ...
st := m.getSlotStates(slot)
for {
// 👇 refresh the states
mustRefresh(ctx, m, st.States)
// 👇 verify the status, return if already FINISHED
if st.States.Status.Load().Is(FINISHED) { return FINISHED }
// 👇 another worker is working on the slot, SKIPPED
if !mustAcquireLock(ctx, m, st.Worker, workerID) { return SKIPPED }
lastID := st.States.LastID
for lastID.Before(startID) {
req := m.QueryRoomEventsRequest{ BeforeID: lastID; Limit: ... }
res := mustRetry(ctx, m.strategy, msgf("query room events"),
func() (QueryRoomEventsResponse, error) {
return m.queryRoomEvents(ctx, req)
}
// 👉 ... save to DynamoDB
// 👉 ... save progress to Redis
lastID = res.LastID
}
}
}
func (m *Migrator) runKeeper(ctx context.Context) {
// ... recover, logs, metrics ...
// ... loop and call runKeeperStep (more on this later) ...
}
func (m *Migrator) runKeeperStep(ctx context.Context) Status {
// ... recover, logs, metrics ...
for {
// 👇 acquire lock and refresh states
// only a single active Keeper across all pods
if !mustAcquireLock(ctx, m, m.keeper) { return SKIPPED }
mustRefresh(ctx, m, m.globalStatus, m.lastSlot)
if m.globalStatus.Load().Is(FINISHED) { return FINISHED }
// 👇 find the last consecutive FINISHED slot
newLastSlot := states.LastSlot
for slot := states.LastSlot; slot.Before(endSlot); slot.Next() {
slotStates := m.getSlotStates(slot)
err := tryRefresh(ctx, m, slotStates)
if err != nil { break }
if !slotStates.Status.Load().Is(FINISHED) { break }
newLastSlot = slot
tryClean(ctx, m, slotStates) // 👈 clean FINISHED slot
}
// 👇 save the mgre:last_slot state
if newLastSlot != states.LastSlot {
mustUpdate(ctx, m, m.lastSlot, newLastSlot)
}
}
}
The plan looks good, right? No, not yet. What happens if any of the above steps fail?
Each Manager, Worker, Keeper run in a goroutine and always retry itself:
func (m *Migrator) initAndRun(ctx context.Context) {
go m.runManager(ctx)
go m.runKeeper(ctx)
for i := 0; i < numWorkers; i++ {
go m.runWorker(ctx)
}
}
func (m *Migrator) runWorker(ctx context.Context) {
defer func() {
r := recover()
if r != nil { log(ctx).Errorf("panic in the outermost layer, will stop") }
}
// 👇 the outermost loop to receive the next slot
// it only contains simple statements to ensure that it never panics
for {
select {
case <- ctx.Done():
return
case slot := <- m.slotCh // 👈 receive slots from channel
s.runTimeslot(ctx, slot) // and execute them one by one
}
}
}
func (m *Migrator) runTimeslot(ctx context.Context) {
defer func() {
r := recover()
if r != nil { log(ctx).Errorf("panic in the outer layer, will stop") }
}
// 👇 the outer loop to retry the migration logic
// it only contains simple statements to ensure that it never panics
t := time.NewTimer(0)
for {
select {
case <-ctx.Done():
return // 👈 stop when the pod stops
case <-t.C:
status := m.runTimeslotStep()
if status.Is(FINISHED, SKIPPED) {
return // 👈 stop when FINISHED or SKIPPED
}
t.Reset(3 * time.Second) // 👈 retry after a few sec
}
}
}
func (m *Migrator) runTimeslotStep(ctx context.Context) (Status) {
defer func() {
r := recover()
if r != nil { logger(ctx, "panic in the inner layer, will retry") }
}
// 👇 the inner loop to execute the migration logic
for {
// ... load states, progress, acquire lock...
// 👇 query database
res, err := retry(ctx, m.strategy, msgf("query room events"),
func() (QueryRoomEventsResponse, error) {
return m.queryRoomEvents(/* ... */)
})
// 👇 even if there is panic, the runManagerStep will recover, stop
// and the outer loop (runManager) will continue retry after few sec
must(err)
// ... save states, progress, refresh lock
}
// ... save status as FINISHED
return FINISHED // 👈 tell the outer loop to stop
}
When any error happens, for example, network timeout:
This ensure that the code always run until all records are migrated, or the pod restarts. In the later case, it will resume the migration progress next time.
API to control the migration:
Stop all using context.WithCancel():
Save progress as states and refresh the lock periodically:
Manager periodically check the mgre:last_slot in Redis:
Log and report the progress:
Monitor the resources and rate limit:
Each Timeslot represents all room events in one hour. We can implement it as a time.Time and store it in Redis as a string with the format 20241020.02.
type Timeslot struct { time.Time }
const slotDuration = time.Hour
func newTimeSlot(t time.Time) Timeslot {
if t.IsZero() { return Timeslot{t} }
// 👉 each slot is an hour
t = t.In(time.UTC).Truncate(kSlotDuration)
return Timeslot{t}
}
func (t Timeslot) String() string {
if t.Time.IsZero() { return "" }
return t.Time.Format("20060102.15")
}
func (t Timeslot) Range() (start, end ulid.ULID) {
return ulid.FromTime(t), ulid.FromTime(t.Add(1))
}
func (t Timeslot) Add(i int) Timeslot {
return Timeslot{t.Time.Add(slotDuration)}
}
func (t Timeslot) Next() Timeslot {
return t.Add(-1) // 👈 we are going from latest to earliest
}
func (t Timeslot) Sub(x Timeslot) int {
return Timeslot{t.Time.Sub(kSlotDuration)}
}
As discussed before, we have 2 loops to handle panic, retry, and resume states. So for this retry() function, we only need to retry the query a couple of times, to be able to tolerate some network failures:
// 👉 this will retry 3 times
strategy := NewSimpleStrategy(
100*time.Millisecond, 200*time.Millisecond, 500*time.Millisecond)
// 👉 call the QueryRoomEvents with retry-ability
func retry(ctx, strategy, msgf("query database"),
func() (QueryRoomEventsRequest, error) {
return m.QueryRoomEvents(ctx, req)
})
// 👉 if all retries failed, stop, and give control back to outer loop
// to try again after a few sec
func mustRetry( /* ... */ ) { /* ... */ }
We can implement a simple retry logic:
func retry[T any](
ctx context.Context, strategy RetryStrategy,
msg fmt.Stringer, fn func() (T, error),
) (T, error) {
for count := 0; ; count++ {
x, err := fn()
if err == nil { return x, err }
if next := strategy.Next(); next > 0 {
logger(ctx).Warnf("failed to %v (attempt %v)", msg, count)
time.Sleep(next)
} else {
logger(ctx).Errorf("failed to %v (attempt %v)", msg, count)
return x, err
}
}
}
An example implementation of retry strategy, which each retry happens after a pre-defined duration:
type RetryStrategy func() time.Duration
func (f RetryStrategy) Next() time.Duration { return f() }
func NewSimpleStrategy(at []time.Duration) RetryStrategy {
i := -1
return func() time.Duration {
i++
if i < len(at) { return i }
return -1; // no more retry
}
}
And implementation of msgf() func which can be used to quickly return a fmt.Stringer:
type StringFunc func() string
func (f StringFunc) String() { return f() }
func msgf(msg string, args ...any) StringFunc {
return func() string {
return fmt.Sprintf(msg, args...)
}
}
There are many states: global status, global progress, slot status, slot progress, locks, etc. For each state, we need to refresh, update, or delete. Each action also needs to be able to retry:
globalStatus := mustRetry(ctx, m.strategy, msgf("load status")
func() (Status, error) {
str, err := m.redisClient.GetString(ctx, "mgre:status")
if err != nil { return 0, err }
if str == "" { return NOT_STARTED }
return parseStatus(str)
})
slotWorker := mustRetry(ctx, m.strategy, msgf("load last slot")
func() (string, error) {
return m.redisClient.GetString(ctx, "mgre:", kSlotWorker(slot))
})
mustRetry(ctx, m.strategy, msg("save status")
func() (int, error) {
str := encodeStatus(newStatus)
err := m.redisClient.SetStringTTL(ctx, "mgre:status", str)
return 0, err
})
func kSlotWorker(slot fmt.Stringer) string {
return fmt.Sprintf("mgre:%v:worker", slot)
}
The code will quickly become too verbose. We can encapsulate the key, including encoding logic, and other configs, in a State struct:
type State[T any] struct {
v atomic.Value
key string
ttl time.Duration
retry RetryStrategy
parse func(string) (T, error)
encode func(T) (string, error)
}
func NewState[T any](
key string, ttl time.Duration, strategy RetryStrategy,
parse func(string) (T, error),
encode func(T) (string, error)
) *State[T] {
var zero T
st := &State[T]{ /* ... */ }
st.v.Store(zero)
return st
}
func (s *State[T]) Load() T {
return s.v.Load().(T)
}
func (s *State[T]) Refresh(ctx context.Context, redis RedisClient) error {
str, err:= retry(ctx, msgf("get key %q", s.key),
func() (string, error) {
return redisClient.Get(ctx, key)
})
if err != nil { return err }
v, err := encode(str)
if err != nil { return err }
s.v.Store(v)
return nil
}
func (s *State[T]) Save(ctx context.Context, redis RedisClient, v T) error {
str, err := s.encode(v)
if err != nil { return err }
return retry(ctx, msgf("set key %q", s.key),
func() (int, error) {
_, err := redisClient.Set(ctx, key, str, s.ttl)
return 0, err
})
}
func (s *State[T]) AcquireLock(ctx context.Context, redis RedisClient, v T) error {
// 👉 similar to Save(), use SetNX instead ...
}
Then implement a few helpers to quickly access them:
type DepsI interface { _redis() RedisClient }
type StateI interface { _key() string; _ttl() time.Duration; /* ... */ }
func mustRefresh(ctx context.Context, deps DepsI, states StateI, msg fmt.Stringer) {
/* ... */
}
func mustSet[T any](ctx context.Context, deps DepsI, state State[T], v T) {
/* ... */
}
Finally, we can simplify the usage:
func (m *Migrator) exampleInit(slot Slot) {
m.globalStatus = NewState(kStatus, longTTL, m.strategy, parseStatus, encodeStatus)
m.slotWorker = NewState(kSlotWorker(slot), shortTTL, m.strategy, parseStr, encodeStr)
// ...
}
func (m *Migrator) exampleRefreshStates() {
mustRefresh(ctx, m, []StateI{m.globalStatus, m.lastSlot}, msgf("load states"))
mustSet(ctx, m, m.globalStatus, FINISHED, msgf("save status"))
// ...
}
That's much better!
The Timeslot struct only contains definition for time slot. We need another struct to encapsulate its states:
type SlotStates struct {
Timeslot // 👉 embedded Timeslot to quickly access methods
status State[Status] // 👉 mgre:TIMESLOT:status
worker State[string] // 👉 mgre:TIMESLOT:worker
}
func newSlotStates(slot Timeslot) *SlotStates {
return &SlotStates{
Timeslot: slot,
status: NewState(kStatus, longTTL, /* ... */),
worker: NewState(kWorker, shortTTL, /* ... */),
}
}
There should be only a single SlotStates for each Timeslot in a pod, shared among manager, keeper, and workers. So it's better to have a centralized place to init and store them:
type Migrator {
// ...
slots map[string]*SlotStates
mu sync.RWMutex
}
func (m *Migrator) getSlotStates(slot Timeslot) *SlotStates {
if st := m._getSlotStates(); st != nil { return st }
m.mu.Lock()
defer m.mu.Unlock()
if st := m.slots[slot.String()]; st != nil { return st }
st := newSlotStates(slot)
s.slots[slot.String()] = st
return st
}
func (m *Migrator) _getSlotStates() *SlotStates {
m.mu.RLock()
defer m.mu.RUnlock()
return m.slots[slot.String()]
}
Phew! That's a lot!! I'm happy that you are still here!
Migrating large volumes of data in a real-world environment is far more complex than a simple script can handle. By carefully designing and implementing the migration code with partitioning, parallel processing, state management, fault tolerance, idempotency, and centralized progress tracking, we can achieve a reliable migration process, maintain data integrity, and minimize downtime.
And sleep well at night too! 😋
I'm Oliver Nguyen -- A software engineer at Connectly.ai. I enjoy learning and seeing a better version of myself each day. Occasionally spin off new open source projects. Share knowledge and thoughts during my journey.
The post is also published at olivernguyen.io.
©️ 2025 All rights reserved. Connectly Inc. Engineered with ❤️, globally.
Company
Let's get started