Forem: Caleb Adewole Erioluwa

Building A Transaction Processing System With MongoDB Trigger

Caleb Adewole Erioluwa — Sun, 19 May 2024 17:15:19 +0000

Event-driven wallet system all piggybacked on top MongoDB atlas service trigger support.
You will learn how triggers in MongoDB work, transaction

handling in MongoDB and little bit of JS
PS - the title was not easy to choose since it manages wallets and transactions

So I am a big fan of designing systems, comparing tradeoffs and implementing systems to see if they work and reduce operation and development costs while still providing value to users.

A while back, during my final exam in college, I was tasked with developing a payment system for the startup I work for. Coming from a company with all the financial resources to run queues, redis clusters e.t.c to a smaller company, We had to pivot.

Disclaimer: My CTO signed off on this article if you were wondering 😂

By pivoting I mean since we typically handle all transactions via third-party providers and the time to market with a new payment system coupled with the legalities involved was not worth it. We decided that we should stick with the third-party approach but manage all the transactions in-house with a wallet system which we didn't have at the time and offload the transaction to the third party which is essentially the banking system

Goals: Keeping it simple and horizontally scalable

By default, one could go with several options to create this system such as

Setup a Queue-based system that updates database information e.g Kafka
Use Event Triggers

One thing about choosing how to design this system is to consider setup and development costs. Since it is a significant setup and development cost to set up a queue system from scratch, we discovered that MongoDB allows us to use something adaptable to our current system. Funny, I forgot to talk about our system.

We work with Sails JS for every backend-related system designed internally and actively use MongoDB as our database engine. The belief is that you can scale horizontally without so much stress with the right system. In summary, we chose event triggers, specifically, Mongo database triggers. What are triggers?

What are Triggers?

According to MongoDB resource documentation, Database triggers can be used to implement complex data interactions. A trigger can update information when related data changes (i.e. when a user profile picture is replaced, triggering an update of the user activity information), or interact with a service when new data is inserted (i.e. when a new calendar entry is added, triggering an email notification).

Main Purposes of Database Triggers

Database triggers serve several important functions:

Audit: Triggers can track changes made to records. For instance, a compliance team might need to know who updated a record. A trigger can automatically record the user who made the change in a "Last Updated By" field.
Data Consistency: Triggers help maintain consistent data formats. For example, to ensure all city names in a database are in uppercase, a trigger can automatically convert any input to uppercase, regardless of how it was entered.
Data Integrity: Triggers enforce rules to ensure valid data combinations. For instance, they can verify that an order's start date is before its end date, preventing invalid date entries before the transaction is finalized.
Data Events: Triggers can initiate actions based on specific events. For example, they can generate a report after data is added or notify users when a certain condition is met, like enough participants joining a game.

MongoDB Triggers

MongoDB Atlas supports database triggers that let the user program the function that will be executed when the database event is fired, take care of server management and provide a handy user interface (which means less code to write).

Types of database triggers supported by MongoDB:

Database Triggers: Perform some action when a document is updated, added, or removed.
Scheduled Triggers: Scheduled actions occur at a specific time or set interval. MongoDB Atlas leverages an intuitive approach to scheduling triggers using the CRON expression language.
Authentication Triggers: Actions happen when creating or deleting a user or logging into MongoDB. This is only supported in the MongoDB realm.

Structure of a Database Trigger On MongoDB

MongoDB Atlas Trigger executes application and database triggers since they are reactive to events or user-defined schedules. Listening to the events of a specific supported and configured type, A trigger is a basic atlas function defined with the signature below when a trigger observes an event that matches your specified configuration fires. The trigger then passes an object representation of the event to the trigger to work with.

exports = async function(changeEvent) {
    // business logic for processing documents based on configuration
}

MongoDB App service provides a way to keep track of the latest execution time for each trigger and guarantees that each event is processed at least once. However, the limitations imposed by the trigger's capacity are determined by its event ordering configuration, which is based on:

Ordered triggers process events from the change stream one at a time in sequence. The next event begins processing only after the previous event finishes processing.
Unordered triggers can process multiple events concurrently, up to 10,000 at once by default. If your Trigger data source is an M10+ Atlas cluster, you can configure individual unordered triggers to exceed the 10,000 concurrent event threshold

Trigger capacity doesn't directly measure throughput or guarantee a specific execution rate. It sets the maximum number of events a Trigger can handle at once. The actual processing rate depends on the Trigger function's runtime logic and the number of events it receives over time. To improve the throughput, One can opt for the following :

Trigger runtime logic optimization
Reduction of event object size with trigger projection filter
Use the right match filter condition to reduce the number of events the trigger processes.

One thing to note is that there is a limitation on the total number of database triggers based on the subscribed cluster size.

Now that we have seen the MongoDB atlas trigger structure, let's take a superficial look at how the wallet system was implemented. Please note that I won't be showing the code just some key functions that help process the event information passed to the database before a commit is made ( company's policy)

The Wallet System

To support any third payment provider, we need our own in-house wallet management system as previous integration showed that not all third-party providers support API that allows users to manage wallet and transaction with lower latency ( milliseconds).

Initially, all our databases were set not to rely on events, so after some engineering debate and R&D., I designed a transaction processing system and a wallet management system.

By setting up a trigger to monitor Transactional event document actions such as creation and deletion, we discovered that handling all necessary resolutions for all supported transactions was possible. The main objective was to differentiate between debit and credit operations. Fortunately, MongoDB facilitates ACID operations through sessions, as an alternative to conventional SQL transactions.

To better understand these, let's take a quick look at the credit function

To have a proper acidic credit logic we had to use session to guarantee the atomic attribute of crediting a user.

I'll also demonstrate a debit function now. By the way, most of the processing is managed through functions, in case you were curious.

Both the credit and debit functions help in adding and removing funds from involved parties and are wrapped around a transaction section which helps enforce the ACID operation.

In the code above, it's important to highlight the transaction option, which informs the database on how to maintain consistency (Transaction API). If an error occurs then the entire commit will be reverted.

Now back to how to trigger functions, We need to extract some information from the parsed change event information from MongoDB. Essentially, a change event is the necessary data pushed from the data to the database but since it is passed via a change stream to the MongoDB database. This change event data is passed into the trigger function based on the configuration.

So every time, the change event matches the set configuration that causes the trigger run.

The outcome of this discussion can be applied to different use cases, such as sending an email when a user registers. In our case, we had to implement it to process payment and manage our user's wallets.

The benefits we gained are the following :

We were able to perform about $65million in volume
Processed ~ $650K in a day ( My CTO showed me cause I didn't believe it initially ) with a capacity supporting ~=50K transactions.
Started building a dashboard on the platform.
Seamless integration
Scalability and offloading to MongoDB atlas service
Minimal to no expense as we operate on free trigger instances.

However, one disadvantage of this solution is all about Vendor locking. Anyway, it is all about risk versus reward.

I hope you gained insight into transactions in MongoDB and how operations can be offloaded to the database layer which can help improve your throughput.

Resources:

- https://www.mongodb.com/docs/manual/core/transactions

- https://www.mongodb.com/docs/atlas/app-services/triggers

- https://www.mongodb.com/resources/products/capabilities/database-triggers

-

I am caleb and you can reach me at Linkedin or follow me on Twitter @Soundboax

Map Reduce - Practical Approach

Caleb Adewole Erioluwa — Sun, 24 Mar 2024 20:25:46 +0000

Prerequisite - A little knowledge about message passing, Golang (maybe but not important )

Mic check !!! 0 1 2.

Yeah I started on a funny index 😅, I hoped I could write this sooner but I have been busy with work. Anyway, let's see how good this note will be ... just kidding.

Map + Reduce = Map Reduce. It's funny how something you have used before was scaled into something big and useful. It's a crazy world of engineering similar to the atomic bomb E=mc^2. In a short while you will understand what I mean.

According to wiki, MapReduce is a programming model and an associated implementation for processing and generating big data sets with a parallel, distributed algorithm on a cluster. It further describes it as a program composed of a mapping procedure, which performs filtering and sorting and a reduce method, which performs a summary operation.

A MapReduce framework is usually composed of three operations (or steps):

Map: each worker node applies the map function to the local data, and writes the output to a temporary storage. A master node ensures that only one copy of the redundant input data is processed. Map(k1,v1) → list(k2,v2)
Shuffle: worker nodes redistribute data based on the output keys (produced by the map function), such that all data belonging to one key is located on the same worker node.
Reduce: worker nodes now process each group of output data, per key, in parallel. Reduce(k2, list (v2)) → list((k3, v3))

A little theoretical background shows that Map operation takes individual values of type A and produces, for each a: A a value b: B.; The Reduce operation requires a binary operation • defined on values of type B; it consists of folding all available b: B to a single value.

From a basic requirements point of view, any MapReduce operation must involve the ability to arbitrarily regroup data being reduced. Such a requirement amounts to two properties of the operation :

associativity: (x • y) • z = x • (y • z)
existence of neutral element e such that e • x = x • e = x for every x:B.

The second property guarantees that, when parallelised over multiple nodes, the nodes that don't have any data to process would have no impact on the result (wiki).

Enough of theory and let's apply the knowledge using a simple example to grasp the concept in a better way. Say we have a list of files with words and we want to know the count of each word in sorted order from all the files. Typically, in a traditional programming way, we will certainly grab the files, read them one after the other and process them i.e count the words using a general map to count the frequency of each word e.g


frequency = {}
files = [ "a.txt","b.txt"]

for file in files:
   with open(file) as f:
      lines = f.readlines()
      for line in lines:
      words = lines.split(" ")
    for word in words:
       if word not in frequency:
         frequency[word] = 1
         continue
            # we are doing map reduce at the same time here 
        freqency[word]+=1 
        # reduce happen during addition here


// blah blah bah

Forgive my unoptimised code lol, I want to depict something close to how it will work. Will this work? yes, it will. Can we make it better using the above map reduce model? Sure we can.

There will be two approaches a coupled one using go channels as a mechanism for parallelising our solution and a decoupled one using RPC. In this article, Let's start with the coupled version

Let's create a map function following this pseudocode. Content extracted from files is sent into the map function and the map function generates a list of key-value data


func Map(name string, document string) []mr.KeyValue {
    // name: document name
        // document: document contents

    // function to detect word separators.
    ff := func(r rune) bool { return !unicode.IsLetter(r) }

    // split contents into an array of words.
    words := strings.FieldsFunc(contents, ff)

    kva := []mr.KeyValue{}
    for _, w := range words {
        kv := mr.KeyValue{w, "1"}
        kva = append(kva, kv)
    }
    return kva
}

For the next part, Let's create a reduce function that processes each group of output data. The reduced functions differ from each other depending on the program's application

func Reduce(key string, values []string) string {
    // return the number of occurrences of this word.
    return strconv.Itoa(len(values))
}

Okay, now that all our important functions have been created ... let's connect the dots (especially the one that makes this framework interesting). Place it in wc.go file

Let's take a step back, Assume we have a coordinator that directs all affairs of our map-reduce activity. Imagine you are in a classroom and your teacher( our coordinator ) decides to give N student documents each to perform a map operation and then tells M students to wait ( not necessarily ) while any result from any of the N students is still processed. Once the result is created from the map function the coordinator forwards it to the reduce function for further action.

Our setup involves a main.go file which loads in the two functions above from their files

func loadPlugin(filename string) (mr.Mapfn, mr.Reducefn, error) {
    p, err := plugin.Open(filename)

    if err != nil {
        return nil, nil, err
    }

    xmapfn, err := p.Lookup("Map")
    if err != nil {
        return nil, nil, fmt.Errorf("cannot look up map function in %s", filename)
    }
    xreducefn, err := p.Lookup("Reduce")
    if err != nil {
        return nil, nil, fmt.Errorf("cannot look up reduce function in %s", filename)
    }
    // assert if the type signature is correct
    mapfn, reducefn := xmapfn.(mr.Mapfn), xreducefn.(mr.Reducefn)

    return mapfn, reducefn, nil
}


func main() {
    pluginFile := flag.String("plugin_file", "./build/fc.so", "name of plugin to run on the application")
    filePath := flag.String("file_path", "./pg-dorian_gray.txt", "file path / glob pattern for files")
    flag.Parse()

    // load function from the passed plugin file 
    mapfn, reducefn, err := loadPlugin(*pluginFile)

    if err != nil {
        log.Fatalf("error loading plugin : %v", err)
    }

    //  clean file path
    *filePath = filepath.Clean(*filePath)
    //  extract the directory path
    dirName := filepath.Dir(*filePath)
    // base filename
    baseFp := filepath.Base(*filePath)

    workloads := []mr.FileInfo{}

    if err := filepath.WalkDir(
        dirName, 
        func(path string, d fs.DirEntry, err error) error {
        if err != nil {
            return fs.SkipDir
        }

        matches, err := filepath.Match(baseFp, filepath.Base(path))

        if err != nil {
            return err
        }

        log.Printf("Comparing baseFp : %s to path :  %s (matches:%v)(regular_file:%v)", baseFp, filepath.Base(path), matches, d.Type().IsRegular())
        if d.Type().IsRegular() && matches {
            fs, err := os.Open(path)
            defer func() {
                fs.Close()
            }()
            if err != nil {
                log.Printf("error occured when opening file path : %v", err)
                return err
            }
            contents, err := io.ReadAll(fs)
            if err != nil {
                log.Printf("error occured when opening file path : %v", err)
                return err
            }
            workloads = append(workloads, mr.FileInfo{
                Filename: filepath.Base(path),
                Contents: contents,
            })
        }
        return nil
    }); err != nil {
        log.Fatalf("error retrieve files : %v", err)
    }

    mw, rw := 2, 2
    coordinator := mr.NewCoordinator(mw, rw).
        RegisterMapFn(mapfn).
        RegisterReduceFn(reducefn).
        LoadWorkloads(workloads)

    startTime := time.Now()
    if err := coordinator.Run(); err != nil {
        log.Fatalf("error occurred while running coordinator : %s", err)
    }
    timeTaken := time.Since(startTime).Milliseconds()
    totalByteProcessed := 0
    for _, fileinfo := range workloads {
        totalByteProcessed += len([]byte(fileinfo.Contents))
    }

    fmt.Printf("Coordinator Runtime [%dms] {mapWorker:%d,reduceWorker:%d} {TotalFileProcessed:%.1fmb} \n", timeTaken, mw, rw, float64(totalByteProcessed/(1024*1024)))
}

So in this code, we did the following :

We load the function written in the go plugin (This was because I wanted to understand how it works back when I was writing it). Check the comment in the code to know where.
List out all files in the directory we want to process so this parses all the file content Slow right but it works (PS - IO is costly🤗). Load the files into the variable workload so we can pass it as an argument to our coordinator.

The coordinator in this case of this example is like our teacher directing every sequence of our actions. And is defined in a coordinator file as shown below.

type Coordinator struct {
    mapWorkers, reduceWorkers int
    taskCh                    chan MTask
    mapCh                     chan []KVPair
    reduceCh                  chan RTask
    resultCh                  chan string
    mapfn                     Mapfn
    reducefn                  Reducefn
    workload                  []FileInfo
    wg                        sync.WaitGroup
    m                         sync.Map
    mapResult                 []KVPair
}

So the coordinator dictates the number of students that will be mapping (as mapWorkers). In this case, the students doing the reduction (a.k.a reduceWorkers). Two channels are created to act as communication mediums for sending and receiving tasks meant for a map operation and reduced operation.

Okay, let's create our coordinator to manage all affairs in a coordinator.go file

func NewCoordinator(numberOfMapWorker int, numberOfReduceWorker int) *Coordinator {
    return &Coordinator{
        mapWorkers:    numberOfMapWorker,
        reduceWorkers: numberOfReduceWorker,
        taskCh:        make(chan MTask, 1),
        resultCh:      make(chan string),
        mapCh:         make(chan []KVPair, 1),
        reduceCh:      make(chan RTask, 1),
        workload:      make([]FileInfo, 0),
        wg:            sync.WaitGroup{},
        m:             sync.Map{},
        mapResult:     []KVPair{},
    }
}

func (c *Coordinator) RegisterMapFn(fn Mapfn) *Coordinator {
    c.mapfn = fn
    return c
}

func (c *Coordinator) RegisterReduceFn(fn Reducefn) *Coordinator {
    c.reducefn = fn
    return c
}

func (c *Coordinator) LoadWorkloads(workloads []FileInfo) *Coordinator {
    c.workload = workloads
    return c
}


func (c *Coordinator) sendOutTaskInformation() {
    c.wg.Add(1)
    go func() {
        defer c.wg.Done()
        defer close(c.taskCh) // Close task channel after sending all tasks
        for id, fileInfo := range c.workload {
            task := MTask{
                ID:      id,
                Payload: fileInfo,
            }
            log.Printf("Sending task for file: %s\n", fileInfo.Filename)
            c.taskCh <- task
        }
    }()
}

func (c *Coordinator) sendReduceOutputForProcessing(mapWorkersWG *sync.WaitGroup) {
    c.wg.Add(1)
    go func() {
        defer c.wg.Done()
        // Use temporary slice to iterate and send reduce tasks
        for kvPairs := range c.mapCh {
            for _, kvPair := range kvPairs {
                fmt.Printf("maps : %s %v\n", kvPair.Key, kvPair.Value)
                c.aggregate(kvPair)
            }
        }

        mapWorkersWG.Wait() // Wait for all map workers to complete

        c.m.Range(func(k, v any) bool {
            ks := k.(string)
            vs := v.([]string)
            log.Printf("Aggregated result: Key: %s, Values: %v\n", ks, len(vs))
            c.reduceCh <- RTask{Key: ks, Value: vs}
            return true
        })

        log.Println("closing c.reduceCh")
        close(c.reduceCh)
    }()
}

func (c *Coordinator) startMapWorkers(mapWorkersWG *sync.WaitGroup, mapDoneCh chan bool) {
    for i := 0; i < c.mapWorkers; i++ {
        id := i + 1
        c.wg.Add(1)
        mapWorkersWG.Add(1)
        go func(id int) {
            defer c.wg.Done()
            defer mapWorkersWG.Done()
            for task := range c.taskCh {
                log.Printf("Map Worker[%d] -> Received task for file: %s\n", id, task.Payload.Filename)
                log.Printf("Map Worker[%d] -> Received task for content: %s\n", id, string(task.Payload.Contents))
                mappedResult := c.mapfn(task.Payload.Filename, string(task.Payload.Contents))
                if mappedResult != nil {
                    c.mapCh <- mappedResult
                    log.Printf("Map Worker[%d] -> Mapped result for file: %s\n", id, task.Payload.Filename)
                } else {
                    log.Printf("Map Worker[%d] -> Error mapping file: %s\n", id, task.Payload.Filename)
                }
            }
            mapDoneCh <- true
            log.Println("Done with map worker", id)
        }(id)
    }
}

func (c *Coordinator) startReduceWorkers(reduceWorkersWG *sync.WaitGroup, reduceDoneCh chan bool) {
    for i := 0; i < c.reduceWorkers; i++ {
        id := i + 1
        c.wg.Add(1)
        reduceWorkersWG.Add(1)
        go func(id int) {
            defer c.wg.Done()
            defer reduceWorkersWG.Done()
            for val := range c.reduceCh {
                log.Printf("Reducer Worker[%d] -> Got %v ", id, val)
                c.resultCh <- c.reducefn(val.Key, val.Value)
            }
            reduceDoneCh <- true
            log.Println("Done with reduce worker", id)
        }(id)
    }
}

func (c *Coordinator) aggregate(v KVPair) {
    existingVal, ok := c.m.Load(v.Key)
    if ok {
        pa := existingVal.([]string)
        pa = append(pa, v.Value)
        c.m.Store(v.Key, pa)
    } else {
        c.m.Store(v.Key, []string{v.Value})
    }
}

func (c *Coordinator) monitorMapAndReduceWorkers(mapDoneCh, reduceDoneCh <-chan bool) {
    c.wg.Add(1)
    go func() {
        defer c.wg.Done()
        for i := c.mapWorkers; i > 0; i-- {
            <-mapDoneCh
        }
        time.Sleep(1000 * time.Millisecond)
        log.Println("closing c.mapCh")
        close(c.mapCh)
    }()
    c.wg.Add(1)
    go func() {
        defer c.wg.Done()
        for i := c.reduceWorkers; i > 0; i-- {
            <-reduceDoneCh
        }
        time.Sleep(10 * time.Millisecond)
        log.Println("closing c.result")
        close(c.resultCh)
    }()
}

func (c *Coordinator) Run() error {

    var (
        mapWorkersWG, reduceWorkersWG sync.WaitGroup
        mapDoneCh                     = make(chan bool, 1) // Signal map completion
        reduceDoneCh                  = make(chan bool, 1) // Signal reduce completion
    )

    // Assign map workers
    c.startMapWorkers(&mapWorkersWG, mapDoneCh)

    // Assign reduce workers
    c.startReduceWorkers(&reduceWorkersWG, reduceDoneCh)

    // Send out task information
    c.sendOutTaskInformation()

    // Send reduce tasks from temporary slice
    c.sendReduceOutputForProcessing(&mapWorkersWG)

    c.monitorMapAndReduceWorkers(mapDoneCh, reduceDoneCh)

    // Write output to file
    oname := "./mr-out-0"
    ofile, err := os.Create(oname)

    if err != nil {
        return err
    }

    defer ofile.Close()

    for result := range c.resultCh {
        if _, err := ofile.WriteString(result + "\n"); err != nil {
            return err
        }
    }

    log.Println("About to close all jobs")

    c.wg.Wait()

    log.Println("Done with all jobs")

    return nil
}

Long right😱? Yup

Let's break down all our functions in order (it is concurrent but we can still think about it detached )

func (c *Coordinator) sendOutTaskInformation() - This function sends out task information to the map workers. It iterates over the workload stored in the coordinator and sends each task (represented by a MTask struct) to the taskCh channel. Each task contains an ID and payload information (file information). The function closes the taskCh channel after sending all tasks.
func (c Coordinator) startMapWorkers(mapWorkersWG sync.WaitGroup, mapDoneCh chan bool)- This function starts the map workers. It creates a specified number of map workers, each running concurrently. Each map worker listens for tasks from the taskCh channel, performs the map function on the received task's file content, and sends the mapped result to the mapCh channel. After completing its tasks, each map worker sends a signal to the mapDoneCh channel to indicate completion.
func (c Coordinator) sendReduceOutputForProcessing(mapWorkersWG sync.WaitGroup) - This function processes the output from map workers for reduction. It receives key-value pairs from the mapCh channel, aggregates them, and sends the aggregated results to the reduceCh channel. Once all map workers are done, it closes the reduceCh channel.
func (c Coordinator) startReduceWorkers(reduceWorkersWG sync.WaitGroup, reduceDoneCh chan bool) - This function starts the reduce workers. Similar to this, it creates a specified number of reduce workers, each running concurrently. Each reduce worker listens for aggregated results from the reduceCh channel, applies the reduce function, and sends the result to the resultCh channel. After completing its tasks, each reduce worker sends a signal to the reduceDoneCh channel to indicate completion.
func (c *Coordinator) aggregate(v KVPair) - This function aggregates key-value pairs. It takes a key-value pair (KVPair), checks if the key already exists in the map (c.m), and appends the value to the existing list associated with the key. If the key does not exist, it creates a new entry in the map with the key and a list containing the value.
func (c *Coordinator) monitorMapAndReduceWorkers(mapDoneCh, reduceDoneCh <-chan bool) - This function monitors the completion of the map and reduces workers. It waits for signals from mapDoneCh and reduceDoneCh channels to know when all map and reduce workers have completed their tasks. Once all map workers are done, it closes the mapCh channel, and once all reduce workers are done, it closes the resultCh channel.

The functions above give you an explanation of what goes on in the code but to make it a bit clear the prior diagram is doing.

To summarise the code aspect of things the snippet below shows the representation of each user type for the problem. The Run() wait till every create routine is done and then collates the result and writes it to a file.

Although this is not a perfect solution to the framework, It will give you an idea of how the RPC solution works.

type KVPair struct {
    Key   string
    Value string
}

type FileInfo struct {
    Filename string
    Contents []byte
}

type Mapfn = func(string, string) []KVPair
type Reducefn = func(string, []string) string

type MTask struct {
    ID      int
    Payload FileInfo
}

type RTask struct {
    Key   string
    Value []string
}

One good thing to note is that the map-reduce model does not necessarily go for speed but instead, it exploits the optimised shuffle operation by separating the Map and Reduce phase. In my case, this is the func (c *Coordinator) aggregate(v KVPair) method.

To better understand this article take a look at the following resource:

https://en.wikipedia.org/wiki/MapReduce

The next article will properly explain how a more practical version of this model can be implemented using RPC. Hopefully, you can check it out 🤗.

Bye for now :).