Implementing Redis Transactions in Go: A Complete Walkthrough

EasyRedis Transaction Implementation

Implements Redis transaction commands MULTI, EXEC, WATCH, UNWATCH, DISCARD with atomicity and optional rollback.

Supported Commands

Multi   : start a transaction
Exec [rb] : execute transaction (optional "rb" triggers automatic rollback on error)
Watch key [key...] : monitor keys; abort if any watched key changes before EXEC
Unwatch : cancel watch
Discard : abort transaction

Atomicity is achieved by taking a lock for the whole transaction and recording original data for rollback.

Project repository: https://github.com/gofish2020/easyredis

Multi – Start Transaction

Marks the current connection as transaction mode; subsequent commands are queued.

func StartMulti(c abstract.Connection) protocol.Reply {
    if c.IsTransaction() {
        return protocol.NewGenericErrReply("multi is already start,do not repeat it")
    }
    c.SetTransaction(true)
    return protocol.NewOkReply()
}

Discard – Cancel Transaction

func DiscardMulti(c abstract.Connection) protocol.Reply {
    if !c.IsTransaction() {
        return protocol.NewGenericErrReply("DISCARD without MULTI")
    }
    c.SetTransaction(false)
    return protocol.NewOkReply()
}

Watch – Monitor Keys

Stores the current version of each specified key; if any version changes before EXEC the transaction aborts.

func Watch(db *DB, conn abstract.Connection, args [][]byte) protocol.Reply {
    if len(args) < 1 {
        return protocol.NewArgNumErrReply("WATCH")
    }
    if conn.IsTransaction() {
        return protocol.NewGenericErrReply("WATCH inside MULTI is not allowed")
    }
    watching := conn.GetWatchKey()
    for _, bkey := range args {
        key := string(bkey)
        watching[key] = db.GetVersion(key) // store current version
    }
    return protocol.NewOkReply()
}

Unwatch – Cancel Monitoring

func UnWatch(db *DB, conn abstract.Connection) protocol.Reply {
    conn.CleanWatchKey()
    return protocol.NewOkReply()
}

Exec – Execute Transaction

Validates transaction state, optional rollback flag, then delegates to db.execMulti.

func ExecMulti(db *DB, conn abstract.Connection, args [][]byte) protocol.Reply {
    if !conn.IsTransaction() {
        return protocol.NewGenericErrReply("EXEC without MULTI")
    }
    defer conn.SetTransaction(false)
    if len(conn.GetTxErrors()) > 0 {
        return protocol.NewGenericErrReply("EXECABORT Transaction discarded because of previous errors.")
    }
    isRollBack := false
    if len(args) > 0 && strings.ToUpper(string(args[0])) == "RB" {
        isRollBack = true
    }
    cmdLines := conn.GetQueuedCmdLine()
    return db.execMulti(conn, cmdLines, isRollBack)
}

execMulti – Core Transaction Logic

Steps performed:

Collect read and write keys for each queued command.

If any watched key version has changed, abort and return an empty multi‑bulk reply.

Lock all involved keys (read‑lock for reads, write‑lock for writes) to guarantee atomicity.

If rollback is enabled, generate undo commands with db.GetUndoLog.

Execute each command under the lock; on error, abort and optionally roll back.

If rollback is required, execute undo commands in reverse order.

For successfully executed write commands, increment their version numbers.

Combine individual replies into a mixed reply and return.

func (db *DB) execMulti(conn abstract.Connection, cmdLines []CmdLine, isRollback bool) protocol.Reply {
    results := make([]protocol.Reply, len(cmdLines))
    versionKeys := make([][]string, len(cmdLines))
    var writeKeys []string
    var readKeys []string
    for idx, cmdLine := range cmdLines {
        cmdName := strings.ToLower(string(cmdLine[0]))
        cmd, ok := commandCenter[cmdName]
        if !ok {
            continue
        }
        readKs, writeKs := cmd.keyFunc(cmdLine[1:])
        readKeys = append(readKeys, readKs...)
        writeKeys = append(writeKeys, writeKs...)
        versionKeys[idx] = append(versionKeys[idx], writeKs...)
    }
    watchingKey := conn.GetWatchKey()
    if isWatchingChanged(db, watchingKey) {
        return protocol.NewEmptyMultiBulkReply()
    }
    db.RWLock(readKeys, writeKeys)
    defer db.RWUnLock(readKeys, writeKeys)

    var undoCmdLines [][]CmdLine
    aborted := false
    for idx, cmdLine := range cmdLines {
        if isRollback {
            undoCmdLines = append(undoCmdLines, db.GetUndoLog(cmdLine))
        }
        reply := db.execWithLock(cmdLine)
        if protocol.IsErrReply(reply) {
            if isRollback {
                undoCmdLines = undoCmdLines[:len(undoCmdLines)-1]
                aborted = true
                break
            }
        }
        results[idx] = reply
    }
    if aborted {
        for i := len(undoCmdLines) - 1; i >= 0; i-- {
            for _, cmd := range undoCmdLines[i] {
                db.execWithLock(cmd)
            }
        }
        return protocol.NewGenericErrReply("EXECABORT Transaction discarded because of previous errors.")
    }
    for idx, keys := range versionKeys {
        if !protocol.IsErrReply(results[idx]) {
            db.addVersion(keys...)
        }
    }
    mixReply := protocol.NewMixReply()
    mixReply.Append(results...)
    return mixReply
}

Demo – No Rollback

Demo – With Rollback

The rollback demo fails because SET key 1 stores a string, while ZADD key 1 member expects a sorted set; the type mismatch triggers an error, causing the transaction to abort and the undo logic to run.