snapshot of https://github.com/golang-migrate/migrate.git (master, 2026-09-15; blobs > 131072B omitted)

database/mongodb/README.md

@@ -0,0 +1,24 @@
+# MongoDB
+
+* Driver work with mongo through [db.runCommands](https://docs.mongodb.com/manual/reference/command/)
+* Migrations support json format. It contains array of commands for `db.runCommand`. Every command is executed in separate request to database 
+* All keys have to be in quotes `"`
+* [Examples](./examples)
+
+# Usage
+
+`mongodb://user:password@host:port/dbname?query` (`mongodb+srv://` also works, but behaves a bit differently. See [docs](https://docs.mongodb.com/manual/reference/connection-string/#dns-seedlist-connection-format) for more information)
+
+| URL Query  | WithInstance Config | Description |
+|------------|---------------------|-------------|
+| `x-migrations-collection` | `MigrationsCollection` | Name of the migrations collection |
+| `x-transaction-mode` | `TransactionMode` | If set to `true` wrap commands in [transaction](https://docs.mongodb.com/manual/core/transactions). Available only for replica set. Driver is using [strconv.ParseBool](https://golang.org/pkg/strconv/#ParseBool) for parsing|
+| `x-advisory-locking` | `true` | Feature flag for advisory locking, if set to false, disable advisory locking |
+| `x-advisory-lock-collection` | `migrate_advisory_lock` | The name of the collection to use for advisory locking.|
+| `x-advisory-lock-timeout` | `15` | The max time in seconds that migrate will wait to acquire a lock before failing. |
+| `x-advisory-lock-timeout-interval` | `10` | The max time in seconds between attempts to acquire the advisory lock, the lock is attempted to be acquired using an exponential backoff algorithm. |
+| `dbname` | `DatabaseName` | The name of the database to connect to |
+| `user` | | The user to sign in as. Can be omitted |
+| `password` | | The user's password. Can be omitted | 
+| `host` | | The host to connect to |
+| `port` | | The port to bind to |

database/mongodb/examples/migrations/001_create_user.down.json

@@ -0,0 +1,5 @@
+[
+  {
+    "dropUser": "deminem"
+  }
+]

database/mongodb/examples/migrations/001_create_user.up.json

@@ -0,0 +1,12 @@
+[
+  {
+    "createUser": "deminem",
+    "pwd": "gogo",
+    "roles": [
+      {
+        "role": "readWrite",
+        "db": "testMigration"
+      }
+    ]
+  }
+]

database/mongodb/examples/migrations/002_create_indexes.down.json

@@ -0,0 +1,10 @@
+[
+  {
+    "dropIndexes": "mycollection",
+    "index": "username_sort_by_asc_created"
+  },
+  {
+    "dropIndexes": "mycollection",
+    "index": "unique_email"
+  }
+]

database/mongodb/examples/migrations/002_create_indexes.up.json

@@ -0,0 +1,21 @@
+[{
+  "createIndexes": "mycollection",
+  "indexes": [
+    {
+      "key": {
+        "username": 1,
+        "created": -1
+      },
+      "name": "username_sort_by_asc_created",
+      "background": true
+    },
+    {
+      "key": {
+        "email": 1
+      },
+      "name": "unique_email",
+      "unique": true,
+      "background": true
+    }
+  ]
+}]

database/mongodb/examples/migrations/003_add_new_field.down.json

@@ -0,0 +1,16 @@
+[
+    {
+        "update": "users",
+        "updates": [
+            {
+                "q": {},
+                "u": {
+                    "$unset": {
+                        "status": ""
+                    }
+                },
+                "multi": true
+            }
+        ]
+    }
+]

database/mongodb/examples/migrations/003_add_new_field.up.json

@@ -0,0 +1,16 @@
+[
+    {
+        "update": "users",
+        "updates": [
+            {
+                "q": {},
+                "u": {
+                    "$set": {
+                        "status": "active"
+                    }
+                },
+                "multi": true
+            }
+        ]
+    }
+]

database/mongodb/examples/migrations/004_replace_field_value_from_another_field.down.json

@@ -0,0 +1,14 @@
+[
+    {
+        "update": "users",
+        "updates": [
+            {
+                "q": {},
+                "u": {
+                    "fullname": ""
+                },
+                "multi": true
+            }
+        ]
+    }
+]

database/mongodb/examples/migrations/004_replace_field_value_from_another_field.up.json

@@ -0,0 +1,23 @@
+[
+    {
+        "aggregate": "users",
+        "pipeline": [
+            {
+                "$project": {
+                    "_id": 1,
+                    "firstname": 1,
+                    "lastname": 1,
+                    "username": 1,
+                    "password": 1,
+                    "email": 1,
+                    "active": 1,
+                    "fullname": { "$concat": ["$firstname", " ", "$lastname"] }
+                }
+            },
+            {
+                "$out": "users"
+            }
+        ],
+        "cursor": {}
+    }
+]

database/mongodb/mongodb.go

@@ -0,0 +1,400 @@
+package mongodb
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/golang-migrate/migrate/v4/database"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/connstring"
+)
+
+func init() {
+	db := Mongo{}
+	database.Register("mongodb", &db)
+	database.Register("mongodb+srv", &db)
+}
+
+var DefaultMigrationsCollection = "schema_migrations"
+
+const DefaultLockingCollection = "migrate_advisory_lock" // the collection to use for advisory locking by default.
+const lockKeyUniqueValue = 0                             // the unique value to lock on. If multiple clients try to insert the same key, it will fail (locked).
+const DefaultLockTimeout = 15                            // the default maximum time to wait for a lock to be released.
+const DefaultLockTimeoutInterval = 10                    // the default maximum intervals time for the locking timout.
+const DefaultAdvisoryLockingFlag = true                  // the default value for the advisory locking feature flag. Default is true.
+const LockIndexName = "lock_unique_key"                  // the name of the index which adds unique constraint to the locking_key field.
+const contextWaitTimeout = 5 * time.Second               // how long to wait for the request to mongo to block/wait for.
+
+var (
+	ErrNoDatabaseName            = fmt.Errorf("no database name")
+	ErrNilConfig                 = fmt.Errorf("no config")
+	ErrLockTimeoutConfigConflict = fmt.Errorf("both x-advisory-lock-timeout-interval and x-advisory-lock-timout-interval were specified")
+)
+
+type Mongo struct {
+	client   *mongo.Client
+	db       *mongo.Database
+	config   *Config
+	isLocked atomic.Bool
+}
+
+type Locking struct {
+	CollectionName string
+	Timeout        int
+	Enabled        bool
+	Interval       int
+}
+type Config struct {
+	DatabaseName         string
+	MigrationsCollection string
+	TransactionMode      bool
+	Locking              Locking
+}
+type versionInfo struct {
+	Version int  `bson:"version"`
+	Dirty   bool `bson:"dirty"`
+}
+
+type lockObj struct {
+	Key       int       `bson:"locking_key"`
+	Pid       int       `bson:"pid"`
+	Hostname  string    `bson:"hostname"`
+	CreatedAt time.Time `bson:"created_at"`
+}
+type findFilter struct {
+	Key int `bson:"locking_key"`
+}
+
+func WithInstance(instance *mongo.Client, config *Config) (database.Driver, error) {
+	if config == nil {
+		return nil, ErrNilConfig
+	}
+	if len(config.DatabaseName) == 0 {
+		return nil, ErrNoDatabaseName
+	}
+	if len(config.MigrationsCollection) == 0 {
+		config.MigrationsCollection = DefaultMigrationsCollection
+	}
+	if len(config.Locking.CollectionName) == 0 {
+		config.Locking.CollectionName = DefaultLockingCollection
+	}
+	if config.Locking.Timeout <= 0 {
+		config.Locking.Timeout = DefaultLockTimeout
+	}
+	if config.Locking.Interval <= 0 {
+		config.Locking.Interval = DefaultLockTimeoutInterval
+	}
+
+	mc := &Mongo{
+		client: instance,
+		db:     instance.Database(config.DatabaseName),
+		config: config,
+	}
+
+	if mc.config.Locking.Enabled {
+		if err := mc.ensureLockTable(); err != nil {
+			return nil, err
+		}
+	}
+	if err := mc.ensureVersionTable(); err != nil {
+		return nil, err
+	}
+
+	return mc, nil
+}
+
+func (m *Mongo) Open(dsn string) (database.Driver, error) {
+	// connstring is experimental package, but it used for parse connection string in mongo.Connect function
+	uri, err := connstring.Parse(dsn)
+	if err != nil {
+		return nil, err
+	}
+	if len(uri.Database) == 0 {
+		return nil, ErrNoDatabaseName
+	}
+	unknown := url.Values(uri.UnknownOptions)
+
+	migrationsCollection := unknown.Get("x-migrations-collection")
+	lockCollection := unknown.Get("x-advisory-lock-collection")
+	transactionMode, err := parseBoolean(unknown.Get("x-transaction-mode"), false)
+	if err != nil {
+		return nil, err
+	}
+	advisoryLockingFlag, err := parseBoolean(unknown.Get("x-advisory-locking"), DefaultAdvisoryLockingFlag)
+	if err != nil {
+		return nil, err
+	}
+	lockingTimout, err := parseInt(unknown.Get("x-advisory-lock-timeout"), DefaultLockTimeout)
+	if err != nil {
+		return nil, err
+	}
+
+	lockTimeoutIntervalValue := unknown.Get("x-advisory-lock-timeout-interval")
+	// The initial release had a typo for this argument but for backwards compatibility sake, we will keep supporting it
+	// and we will error out if both values are set.
+	lockTimeoutIntervalValueFromTypo := unknown.Get("x-advisory-lock-timout-interval")
+
+	lockTimeout := lockTimeoutIntervalValue
+
+	if lockTimeoutIntervalValue != "" && lockTimeoutIntervalValueFromTypo != "" {
+		return nil, ErrLockTimeoutConfigConflict
+	} else if lockTimeoutIntervalValueFromTypo != "" {
+		lockTimeout = lockTimeoutIntervalValueFromTypo
+	}
+
+	maxLockCheckInterval, err := parseInt(lockTimeout, DefaultLockTimeoutInterval)
+
+	if err != nil {
+		return nil, err
+	}
+	client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(dsn))
+	if err != nil {
+		return nil, err
+	}
+
+	if err = client.Ping(context.TODO(), nil); err != nil {
+		return nil, err
+	}
+	mc, err := WithInstance(client, &Config{
+		DatabaseName:         uri.Database,
+		MigrationsCollection: migrationsCollection,
+		TransactionMode:      transactionMode,
+		Locking: Locking{
+			CollectionName: lockCollection,
+			Timeout:        lockingTimout,
+			Enabled:        advisoryLockingFlag,
+			Interval:       maxLockCheckInterval,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return mc, nil
+}
+
+// Parse the url param, convert it to boolean
+// returns error if param invalid. returns defaultValue if param not present
+func parseBoolean(urlParam string, defaultValue bool) (bool, error) {
+
+	// if parameter passed, parse it (otherwise return default value)
+	if urlParam != "" {
+		result, err := strconv.ParseBool(urlParam)
+		if err != nil {
+			return false, err
+		}
+		return result, nil
+	}
+
+	// if no url Param passed, return default value
+	return defaultValue, nil
+}
+
+// Parse the url param, convert it to int
+// returns error if param invalid. returns defaultValue if param not present
+func parseInt(urlParam string, defaultValue int) (int, error) {
+
+	// if parameter passed, parse it (otherwise return default value)
+	if urlParam != "" {
+		result, err := strconv.Atoi(urlParam)
+		if err != nil {
+			return -1, err
+		}
+		return result, nil
+	}
+
+	// if no url Param passed, return default value
+	return defaultValue, nil
+}
+func (m *Mongo) SetVersion(version int, dirty bool) error {
+	migrationsCollection := m.db.Collection(m.config.MigrationsCollection)
+	if err := migrationsCollection.Drop(context.TODO()); err != nil {
+		return &database.Error{OrigErr: err, Err: "drop migrations collection failed"}
+	}
+	_, err := migrationsCollection.InsertOne(context.TODO(), bson.M{"version": version, "dirty": dirty})
+	if err != nil {
+		return &database.Error{OrigErr: err, Err: "save version failed"}
+	}
+	return nil
+}
+
+func (m *Mongo) Version() (version int, dirty bool, err error) {
+	var versionInfo versionInfo
+	err = m.db.Collection(m.config.MigrationsCollection).FindOne(context.TODO(), bson.M{}).Decode(&versionInfo)
+	switch {
+	case err == mongo.ErrNoDocuments:
+		return database.NilVersion, false, nil
+	case err != nil:
+		return 0, false, &database.Error{OrigErr: err, Err: "failed to get migration version"}
+	default:
+		return versionInfo.Version, versionInfo.Dirty, nil
+	}
+}
+
+func (m *Mongo) Run(migration io.Reader) error {
+	migr, err := io.ReadAll(migration)
+	if err != nil {
+		return err
+	}
+	var cmds []bson.D
+	err = bson.UnmarshalExtJSON(migr, true, &cmds)
+	if err != nil {
+		return fmt.Errorf("unmarshaling json error: %s", err)
+	}
+	if m.config.TransactionMode {
+		if err := m.executeCommandsWithTransaction(context.TODO(), cmds); err != nil {
+			return err
+		}
+	} else {
+		if err := m.executeCommands(context.TODO(), cmds); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *Mongo) executeCommandsWithTransaction(ctx context.Context, cmds []bson.D) error {
+	err := m.db.Client().UseSession(ctx, func(sessionContext mongo.SessionContext) error {
+		if err := sessionContext.StartTransaction(); err != nil {
+			return &database.Error{OrigErr: err, Err: "failed to start transaction"}
+		}
+		if err := m.executeCommands(sessionContext, cmds); err != nil {
+			// When command execution is failed, it's aborting transaction
+			// If you tried to call abortTransaction, it`s return error that transaction already aborted
+			return err
+		}
+		if err := sessionContext.CommitTransaction(sessionContext); err != nil {
+			return &database.Error{OrigErr: err, Err: "failed to commit transaction"}
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *Mongo) executeCommands(ctx context.Context, cmds []bson.D) error {
+	for _, cmd := range cmds {
+		err := m.db.RunCommand(ctx, cmd).Err()
+		if err != nil {
+			return &database.Error{OrigErr: err, Err: fmt.Sprintf("failed to execute command:%v", cmd)}
+		}
+	}
+	return nil
+}
+
+func (m *Mongo) Close() error {
+	return m.client.Disconnect(context.TODO())
+}
+
+func (m *Mongo) Drop() error {
+	return m.db.Drop(context.TODO())
+}
+
+func (m *Mongo) ensureLockTable() error {
+	indexes := m.db.Collection(m.config.Locking.CollectionName).Indexes()
+
+	indexOptions := options.Index().SetUnique(true).SetName(LockIndexName)
+	_, err := indexes.CreateOne(context.TODO(), mongo.IndexModel{
+		Options: indexOptions,
+		Keys:    findFilter{Key: -1},
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// ensureVersionTable checks if versions table exists and, if not, creates it.
+// Note that this function locks the database, which deviates from the usual
+// convention of "caller locks" in the MongoDb type.
+func (m *Mongo) ensureVersionTable() (err error) {
+	if err = m.Lock(); err != nil {
+		return err
+	}
+
+	defer func() {
+		if e := m.Unlock(); e != nil {
+			err = errors.Join(err, e)
+		}
+	}()
+
+	if err != nil {
+		return err
+	}
+	if _, _, err = m.Version(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// Utilizes advisory locking on the config.LockingCollection collection
+// This uses a unique index on the `locking_key` field.
+func (m *Mongo) Lock() error {
+	return database.CasRestoreOnErr(&m.isLocked, false, true, database.ErrLocked, func() error {
+		if !m.config.Locking.Enabled {
+			return nil
+		}
+
+		pid := os.Getpid()
+		hostname, err := os.Hostname()
+		if err != nil {
+			hostname = fmt.Sprintf("Could not determine hostname. Error: %s", err.Error())
+		}
+
+		newLockObj := lockObj{
+			Key:       lockKeyUniqueValue,
+			Pid:       pid,
+			Hostname:  hostname,
+			CreatedAt: time.Now(),
+		}
+		operation := func() error {
+			timeout, cancelFunc := context.WithTimeout(context.Background(), contextWaitTimeout)
+			_, err := m.db.Collection(m.config.Locking.CollectionName).InsertOne(timeout, newLockObj)
+			defer cancelFunc()
+			return err
+		}
+		exponentialBackOff := backoff.NewExponentialBackOff()
+		duration := time.Duration(m.config.Locking.Timeout) * time.Second
+		exponentialBackOff.MaxElapsedTime = duration
+		exponentialBackOff.MaxInterval = time.Duration(m.config.Locking.Interval) * time.Second
+
+		err = backoff.Retry(operation, exponentialBackOff)
+		if err != nil {
+			return database.ErrLocked
+		}
+
+		return nil
+	})
+}
+
+func (m *Mongo) Unlock() error {
+	return database.CasRestoreOnErr(&m.isLocked, true, false, database.ErrNotLocked, func() error {
+		if !m.config.Locking.Enabled {
+			return nil
+		}
+
+		filter := findFilter{
+			Key: lockKeyUniqueValue,
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), contextWaitTimeout)
+		_, err := m.db.Collection(m.config.Locking.CollectionName).DeleteMany(ctx, filter)
+		defer cancel()
+
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+}

database/mongodb/mongodb_test.go

@@ -0,0 +1,425 @@
+package mongodb
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+
+	"log"
+
+	"github.com/golang-migrate/migrate/v4"
+	"io"
+	"os"
+	"strconv"
+	"testing"
+	"time"
+)
+
+import (
+	"github.com/dhui/dktest"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+import (
+	dt "github.com/golang-migrate/migrate/v4/database/testing"
+	"github.com/golang-migrate/migrate/v4/dktesting"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
+)
+
+var (
+	opts = dktest.Options{PortRequired: true, ReadyFunc: isReady}
+	// Supported versions: https://www.mongodb.com/support-policy
+	specs = []dktesting.ContainerSpec{
+		{ImageName: "mongo:5.0", Options: opts},
+		{ImageName: "mongo:6.0", Options: opts},
+		{ImageName: "mongo:7.0", Options: opts},
+		{ImageName: "mongo:8.0", Options: opts},
+	}
+)
+
+func mongoConnectionString(host, port string) string {
+	// there is connect option for excluding serverConnection algorithm
+	// it's let avoid errors with mongo replica set connection in docker container
+	return fmt.Sprintf("mongodb://%s:%s/testMigration?connect=direct", host, port)
+}
+
+func isReady(ctx context.Context, c dktest.ContainerInfo) bool {
+	ip, port, err := c.FirstPort()
+	if err != nil {
+		return false
+	}
+
+	client, err := mongo.Connect(ctx, options.Client().ApplyURI(mongoConnectionString(ip, port)))
+	if err != nil {
+		return false
+	}
+	defer func() {
+		if err := client.Disconnect(ctx); err != nil {
+			log.Println("close error:", err)
+		}
+	}()
+
+	if err = client.Ping(ctx, nil); err != nil {
+		switch err {
+		case io.EOF:
+			return false
+		default:
+			log.Println(err)
+		}
+		return false
+	}
+	return true
+}
+
+func Test(t *testing.T) {
+	t.Run("test", test)
+	t.Run("testMigrate", testMigrate)
+	t.Run("testWithAuth", testWithAuth)
+	t.Run("testLockWorks", testLockWorks)
+
+	t.Cleanup(func() {
+		for _, spec := range specs {
+			t.Log("Cleaning up ", spec.ImageName)
+			if err := spec.Cleanup(); err != nil {
+				t.Error("Error removing ", spec.ImageName, "error:", err)
+			}
+		}
+	})
+}
+
+func test(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.FirstPort()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addr := mongoConnectionString(ip, port)
+		p := &Mongo{}
+		d, err := p.Open(addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		dt.TestNilVersion(t, d)
+		dt.TestLockAndUnlock(t, d)
+		dt.TestRun(t, d, bytes.NewReader([]byte(`[{"insert":"hello","documents":[{"wild":"world"}]}]`)))
+		dt.TestSetVersion(t, d)
+		dt.TestDrop(t, d)
+	})
+}
+
+func testMigrate(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.FirstPort()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addr := mongoConnectionString(ip, port)
+		p := &Mongo{}
+		d, err := p.Open(addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		m, err := migrate.NewWithDatabaseInstance("file://./examples/migrations", "", d)
+		if err != nil {
+			t.Fatal(err)
+		}
+		dt.TestMigrate(t, m)
+	})
+}
+
+func testWithAuth(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.FirstPort()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addr := mongoConnectionString(ip, port)
+		p := &Mongo{}
+		d, err := p.Open(addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		createUserCMD := []byte(`[{"createUser":"deminem","pwd":"gogo","roles":[{"role":"readWrite","db":"testMigration"}]}]`)
+		err = d.Run(bytes.NewReader(createUserCMD))
+		if err != nil {
+			t.Fatal(err)
+		}
+		testcases := []struct {
+			name            string
+			connectUri      string
+			isErrorExpected bool
+		}{
+			{"right auth data", "mongodb://deminem:gogo@%s:%v/testMigration", false},
+			{"wrong auth data", "mongodb://wrong:auth@%s:%v/testMigration", true},
+		}
+
+		for _, tcase := range testcases {
+			t.Run(tcase.name, func(t *testing.T) {
+				mc := &Mongo{}
+				d, err := mc.Open(fmt.Sprintf(tcase.connectUri, ip, port))
+				if err == nil {
+					defer func() {
+						if err := d.Close(); err != nil {
+							t.Error(err)
+						}
+					}()
+				}
+
+				switch {
+				case tcase.isErrorExpected && err == nil:
+					t.Fatalf("no error when expected")
+				case !tcase.isErrorExpected && err != nil:
+					t.Fatalf("unexpected error: %v", err)
+				}
+			})
+		}
+	})
+}
+
+func testLockWorks(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.FirstPort()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addr := mongoConnectionString(ip, port)
+		p := &Mongo{}
+		d, err := p.Open(addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		dt.TestRun(t, d, bytes.NewReader([]byte(`[{"insert":"hello","documents":[{"wild":"world"}]}]`)))
+
+		mc := d.(*Mongo)
+
+		err = mc.Lock()
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = mc.Unlock()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		err = mc.Lock()
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = mc.Unlock()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// enable locking,
+		//try to hit a lock conflict
+		mc.config.Locking.Enabled = true
+		mc.config.Locking.Timeout = 1
+		err = mc.Lock()
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = mc.Lock()
+		if err == nil {
+			t.Fatal("should have failed, mongo should be locked already")
+		}
+	})
+}
+
+func TestTransaction(t *testing.T) {
+	transactionSpecs := []dktesting.ContainerSpec{
+		{ImageName: "mongo:4", Options: dktest.Options{PortRequired: true, ReadyFunc: isReady,
+			Cmd: []string{"mongod", "--bind_ip_all", "--replSet", "rs0"}}},
+	}
+	t.Cleanup(func() {
+		for _, spec := range transactionSpecs {
+			t.Log("Cleaning up ", spec.ImageName)
+			if err := spec.Cleanup(); err != nil {
+				t.Error("Error removing ", spec.ImageName, "error:", err)
+			}
+		}
+	})
+
+	dktesting.ParallelTest(t, transactionSpecs, func(t *testing.T, c dktest.ContainerInfo) {
+		ip, port, err := c.FirstPort()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(mongoConnectionString(ip, port)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = client.Ping(context.TODO(), nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		//rs.initiate()
+		err = client.Database("admin").RunCommand(context.TODO(), bson.D{bson.E{Key: "replSetInitiate", Value: bson.D{}}}).Err()
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = waitForReplicaInit(client)
+		if err != nil {
+			t.Fatal(err)
+		}
+		d, err := WithInstance(client, &Config{
+			DatabaseName: "testMigration",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.Close(); err != nil {
+				t.Error(err)
+			}
+		}()
+		//We have to create collection
+		//transactions don't support operations with creating new dbs, collections
+		//Unique index need for checking transaction aborting
+		insertCMD := []byte(`[
+				{"create":"hello"},
+				{"createIndexes": "hello",
+					"indexes": [{
+						"key": {
+							"wild": 1
+						},
+						"name": "unique_wild",
+						"unique": true,
+						"background": true
+					}]
+			}]`)
+		err = d.Run(bytes.NewReader(insertCMD))
+		if err != nil {
+			t.Fatal(err)
+		}
+		testcases := []struct {
+			name            string
+			cmds            []byte
+			documentsCount  int64
+			isErrorExpected bool
+		}{
+			{
+				name: "success transaction",
+				cmds: []byte(`[{"insert":"hello","documents":[
+										{"wild":"world"},
+										{"wild":"west"},
+										{"wild":"natural"}
+									 ]
+								  }]`),
+				documentsCount:  3,
+				isErrorExpected: false,
+			},
+			{
+				name: "failure transaction",
+				//transaction have to be failure - duplicate unique key wild:west
+				//none of the documents should be added
+				cmds: []byte(`[{"insert":"hello","documents":[{"wild":"flower"}]},
+									{"insert":"hello","documents":[
+										{"wild":"cat"},
+										{"wild":"west"}
+									 ]
+								  }]`),
+				documentsCount:  3,
+				isErrorExpected: true,
+			},
+		}
+		for _, tcase := range testcases {
+			t.Run(tcase.name, func(t *testing.T) {
+				client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(mongoConnectionString(ip, port)))
+				if err != nil {
+					t.Fatal(err)
+				}
+				err = client.Ping(context.TODO(), nil)
+				if err != nil {
+					t.Fatal(err)
+				}
+				d, err := WithInstance(client, &Config{
+					DatabaseName:    "testMigration",
+					TransactionMode: true,
+				})
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer func() {
+					if err := d.Close(); err != nil {
+						t.Error(err)
+					}
+				}()
+				runErr := d.Run(bytes.NewReader(tcase.cmds))
+				if runErr != nil {
+					if !tcase.isErrorExpected {
+						t.Fatal(runErr)
+					}
+				}
+				documentsCount, err := client.Database("testMigration").Collection("hello").CountDocuments(context.TODO(), bson.M{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if tcase.documentsCount != documentsCount {
+					t.Fatalf("expected %d and actual %d documents count not equal. run migration error:%s", tcase.documentsCount, documentsCount, runErr)
+				}
+			})
+		}
+	})
+}
+
+type isMaster struct {
+	IsMaster bool `bson:"ismaster"`
+}
+
+func waitForReplicaInit(client *mongo.Client) error {
+	ticker := time.NewTicker(time.Second * 1)
+	defer ticker.Stop()
+	timeout, err := strconv.Atoi(os.Getenv("MIGRATE_TEST_MONGO_REPLICA_SET_INIT_TIMEOUT"))
+	if err != nil {
+		timeout = 30
+	}
+	timeoutTimer := time.NewTimer(time.Duration(timeout) * time.Second)
+	defer timeoutTimer.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			var status isMaster
+			//Check that node is primary because
+			//during replica set initialization, the first node first becomes a secondary and then becomes the primary
+			//should consider that initialization is completed only after the node has become the primary
+			result := client.Database("admin").RunCommand(context.TODO(), bson.D{bson.E{Key: "isMaster", Value: 1}})
+			if err := result.Decode(&status); err != nil {
+				return err
+			}
+			if status.IsMaster {
+				return nil
+			}
+		case <-timeoutTimer.C:
+			return fmt.Errorf("replica init timeout")
+		}
+	}
+
+}