Stateful flow: organising distributed processing in golang

Here I would like to develop some thoughts regarding state management raised in the previous article. In functional programming it's customary explicitly designate the state. However, that could lead to so-called state explosion. Sometimes it would be better to remain some general state implicit and to operate substate sets.

Consider real life case. User orders some product. What is implemented as a http request to our service. Processing this inquiry we need to make request to an accounting service to create record and to check if product is available and another request to a financials service to provide payment. Of course, we need store the order to a database. All of this external services return successive or not response. Which leads to updating requests to other services. For example, if financials service returned out of balance status, we should update the record in accounting service as rejected and free reserved product and update order status in the database.

Thus, there are three units: storage, bookkeeping and financials. Each module has two types of functionality: applying order to the corresponding service and updating status. They form the flow. Units consequently message each other. Each module has its own state depending on the response from external service. We might create a general status corresponding to each unit state. For example BookkeepingSuccess, BookkeepingProductNotAvailable, BookkeepingInternalError and analogical for others. But the whole web of units can be in two states: success or proceed and cancel. These states are translated to units' actions. Responses from services are transformed to flow status in their turn. If bookkeeping service returned that product was not available, then the status of flow becomes Cancel and messages are spread propagating this state to modules. We don't need to store the flow state explicitly, we can just use two types of signals between processing modules. So we got the stateful flow.

Some possible scenarios are showed in the diagram

Take a look at the sample application. The information processing flow itself is represented as OrderFlow struct

type OrderStatus string

const (
  OrderCreated             OrderStatus = "created"
  OrderSuccess             OrderStatus = "success"
  OrderInternalError       OrderStatus = "internal_error"
  OrderProductNotAvailable OrderStatus = "product_not_available"
  OrderBalanceNotEnough    OrderStatus = "balance_not_enough"
)

type OrderFlow struct {
  Config       *Config
  OrderStatus  OrderStatus
  OrderID      int
  UserID       uuid.UUID
  ProductID    int
  ProductPrice decimal.Decimal
}

This struct stores order information and status which we return in the answer to user. Units intercommunicate by StatusStream struct

type FlowStatus int

const (
  Proceed FlowStatus = 1
  Cancel  FlowStatus = 2
)

type StatusStream struct {
  Forward chan FlowStatus
  Back    chan FlowStatus
}

Module generally takes two StatusStream struct. The first provides connection with previous module and the second one connects with the next module. By Forward channel unit sends status message to next unit and by Back channel receives message from next unit.

Consider bookkeeping unit

type BookkepingUnit model.OrderFlow

func (f *BookkepingUnit) Process(previous *model.StatusStream, next *model.StatusStream) {
  status := <-previous.Forward
  if status == model.Cancel {
    next.Forward <- model.Cancel
    return
  }

...

resp, err := general.MakeHTTPRequest[ApplyRequest, ApplyResponse]("POST", f.Config.BookkepingApplyURL, &breq)
switch resp.Result.Status {
  case Success:
    next.Forward <- model.Proceed
  case ProductNotAvailable:
    f.OrderStatus = model.OrderProductNotAvailable
    next.Forward <- model.Cancel
    previous.Back <- model.Cancel
    go logger.LogUnit(logger.Info, f.Config.Name, nil,
      f.OrderID, unit, string(ProductNotAvailable))
    return
}

...

status = <-next.Back
if status == model.Cancel {
  previous.Back <- model.Cancel
  updateStatus = OrderCanceled
}
updresp, err := general.MakeHTTPRequest[interface{}, UpdateResponse]("PUT", url, nil)
if updresp.Status == general.StatusError {
  go logger.LogUnit(logger.Info, f.Config.Name, nil,
    f.OrderID, unit, string(model.OrderInternalError))
  if status == model.Proceed {
    next.Forward <- model.Cancel
    previous.Back <- model.Cancel
  }
}

Modules inherited from OrderFlow, for logical separation the type redefinition is used. At first, unit waits for signal from previous stage. Then it makes a request to external service. In case successive response the Proceed signal is sent to the next unit. If response was considered as unsuccessful, for example, no products were left, the Cancel signal propagates to previous and next modules. Depending on the status returned from sequent stage, the update request is made with OrderPaid or OrderCanceled status. If we got an error when updating record status, we also cancel the order and send Cancel signal to modules.

Launching the flow is made in the flow.Process function

func Process(flow *model.OrderFlow) {
  repo := storage.NewRepository(general.NewPgsql(flow.Config.DBConnectionString))
  bu := (*bookkeeping.BookkepingUnit)(flow)
  fu := (*financials.FinancialsUnit)(flow)
  su := (*storage.StorageUnit)(flow)
  start := model.NewStatusStream()
  storage2bookkeeping := model.NewStatusStream()
  bookkeeping2financials := model.NewStatusStream()
  go su.Process(repo, start, storage2bookkeeping)
  go bu.Process(storage2bookkeeping, bookkeeping2financials)
  go fu.Process(bookkeeping2financials)
  start.Forward <- model.Proceed
  <-start.Back
}

Here we create units and channels, start the processing and wait for the processing finish. Also note the nice declarative style of this operations.
In the endpoint layer the pre- and post-processing are performed.

func MakeOrderEndpoint(cfg *model.Config) general.Endpoint {
  return func(w http.ResponseWriter, r *http.Request) {
    w.Header().Add("Content-Type", "application/json")
    req, err := general.RequestDecode[model.OrderRequest](r)

...

    f := &model.OrderFlow{
      Config:       cfg,
      UserID:       req.UserID,
      ProductID:    req.ProductID,
      ProductPrice: req.ProductPrice,
    }
    flow.Process(f)
    res := model.OrderResult{
      OrderStatus: f.OrderStatus,
      OrderID:     f.OrderID,
    }
    resp := model.OrderResponse{
      Status: general.StatusOK,
      Result: &res,
    }

...

    json.NewEncoder(w).Encode(resp)

In this layer we decode the user order request, form OrderfFlow object, launch the flow, and make dto and return the answer to the user.
We need to message user the order's state. So in each unit some status algebra is made. Service responces are translated to order status. In finacials unit for example

switch resp.Result.Status {
  case Success:
    transactionID = *resp.Result.TransactionID
    previous.Back <- model.Proceed
  case BalanceNotEnough:
    f.OrderStatus = model.OrderBalanceNotEnough
    go logger.LogUnit(logger.Info, f.Config.Name, nil,
      f.OrderID, unit, string(model.OrderBalanceNotEnough))
    previous.Back <- model.Cancel
}

In modules we log errors so we can reconstruct events if something went wrong.

Golang has powerful tools for distributed information processing. I proposed here the stateful flow conception. It can branch out and become stateful web. Presented toolkit could help with this situation as well.