Skip to content

API Reference#

Tip

All the functionality of zio-maelstrom is available via following import statement

import com.bilalfazlani.zioMaelstrom.*

Primitive Types#

NodeId#

NodeId represents a unique identifier for a node. It is a wrapper (opaque type) around String and can be created using NodeId(String) method.

MessageId#

MessageId represents a unique identifier for a message. It is a wrapper (opaque type) around Int and can be created using MessageId(Int) method.

Protocol#

In order to send and receive messages from a node, we need to create case classes that contain the fields which we need to send or receive. The case classes dont need to contain standard fields defined in maelstrom protocol like src, dest, type, msg_id, in_reply_to, etc. Those are automatically added and parsed by the runtime.

case class Echo(text: String) derives JsonCodec

{
    "src" : "c1", 
    "dest" : "n1", 
    "body" : {
        "type": "echo", 
        "msg_id": 10, 
        "in_reply_to": 5, 
        "text": "Hello" 
    }
}

val echo = Echo("Hello")
val nodeId = NodeId("n1")
nodeId.send(echo)

{
    "src" : "c1"
    "dest" : "n1",
    "body" : {
        "type": "echo",
        "text": "Hello"
    }
}

val echo = Echo("Hello")
val nodeId = NodeId("n1")
nodeId.ask[Echo](echo, 5.seconds)

{
    "src" : "c1",
    "dest" : "n1",
    "body" : {
        "type": "echo",
        "text": "Hello",
        "msg_id": 10
    }
}

val echo = Echo("Hello")
reply(echo, 3.seconds)

{
    "src" : "n1",
    "dest" : "c1",
    "body" : {
        "type": "echo",
        "text": "Hello",
        "in_reply_to": 10
    }
}

Note

Keep in mind that if you use ask api, framework adds a msg_id to the message. If you use reply api, framework adds a in_reply_to to the message.

Caution

If you try to use reply api for a message that does not have a msg_id (i.e. sent using send api), it will throw an error at runtime.

Caution

If you use ask api, the called will wait for the reply with a timeout. If the reply is not received within the timeout, it will return Timeout error.

The idea behind framework design is that when writing solutions to problems, should should not have to think about msg_id and other things. Much like when we write an HTTP/GRPC client or server.

I/O APIs#

1. receive#

receive api takes a handler function I => ZIO[MaelstromRuntime & R, Nothing, Unit]

Note

  1. I needs have a zio.json.JsonDecoder instance
  2. R can be anything. You will need to provide R & MaelstromRuntime when you run the ZIO effect

Here's an example

Receive
case class Gossip(numbers: Seq[Int]) derives JsonCodec

val messageHandler =
  receive[Gossip] (msg =>
      for {
        src <- MaelstromRuntime.src
        me <- MaelstromRuntime.me
        others <- MaelstromRuntime.others
        _ <- ZIO.logDebug(s"received $msg from $src") 
        _ <- ZIO.logDebug(s"my node id is $me") 
        _ <- ZIO.logDebug(s"other node ids are $others") 
      } yield ()
  )

2. send#

You can send a message to any NodeId using NodeId.send() API. It takes a instance of a case class which has a zio.json.JsonEncoder.

Send
case class Gossip(numbers: Seq[Int]) derives JsonCodec

val messageHandler =
  receive[Gossip] (_ =>
    for
      others <- MaelstromRuntime.others
      _ <- ZIO.foreach(others)(_.send(Gossip(Seq(1,2)))).unit 
    yield ()
  )

val result = NodeId("n5") send Gossip(Seq(1,2))

3. ask#

ask api is a combination of send and receive. It sends a message to a remote node and waits for a reply. It also takes a timeout argument which is the maximum time to wait for a reply. It expects a zio.json.JsonDecoder instance for the reply & a zio.json.JsonEncoder instance for the request message.

Ask
case class Gossip(numbers: Seq[Int]) derives JsonCodec

case class GossipOk(myNumbers: Seq[Int]) derives JsonCodec

val gossipResult: ZIO[MaelstromRuntime, AskError, GossipOk] =
  NodeId("n2").ask[GossipOk](Gossip(Seq(1,2)), 5.seconds)

The ask api can return either a successful response or an AskError

AskError
type AskError = Error | DecodingFailure | Timeout

Ask error can be one of the following:

  1. Timeout if the reply was not received within given duration
  2. DecodingFailure if the reply could not be decoded into the given type
  3. Error if the sender sends an error message instead of the reply message.
Ask error handling
case class Query(id: Int) derives JsonCodec

case class Answer(text: String) derives JsonCodec

val askResponse: ZIO[MaelstromRuntime, AskError, Unit] = for
  answer <- NodeId("g4").ask[Answer](Query(1), 5.seconds)
  _      <- ZIO.logInfo(s"answer: $answer")
yield ()

askResponse
  .catchAll {
    case t: Timeout         => ZIO.logError(s"timeout: ${t.timeout}")
    case d: DecodingFailure => ZIO.logError(s"decoding failure: ${d.error}")
    case e: Error           =>
      val code: ErrorCode = e.code
      val text: String    = e.text
      ZIO.logError(s"error code: $code, error text: $text")
  }

Sender can send an error message if it encounters an error while processing the request message or when request is invalid. You can read more about error messages in the error messages section

4. reply#

You can call reply api to send a reply message to the source of the current message (if the message was sent using ask api)

Reply
case class Gossip(numbers: Seq[Int]) derives JsonCodec

case class GossipOk(myNumbers: Seq[Int]) derives JsonCodec

val messageHandler = receive[Gossip] (_ => reply(GossipOk(Seq(1,2))))

reply api takes an instance of a case class which has a zio.json.JsonEncoder

Context APIs#

MaelstromRuntime.me

returns the NodeId of the current node

Signature: ZIO[MaelstromRuntime, Nothing, NodeId]

MaelstromRuntime.others

returns a list of NodeId of all other nodes in the cluster

Signature: ZIO[MaelstromRuntime, Nothing, Set[NodeId]]

MaelstromRuntime.src

returns the NodeId of the source of the current message

Signature: ZIO[MessageContext, Nothing, NodeId]

Error messages#

zio-maelstrom has a built in data type for error messages called Error

Error
case class Error(
    code: ErrorCode,
    text: String
) derives JsonCodec

It supports all the standard maelstrom error codes as well as ability to send custom error codes

View all error codes
Error codes
sealed trait ErrorCode(private val code1: Int, val definite: Boolean) {
  def code: Int = code1
  override def toString: String = s"error: ${this.getClass.getSimpleName.replace("$","")}, code: $code"
}

object ErrorCode:
  /**
    * Indicates that the requested operation could not be completed within a timeout.
    */
  object Timeout extends ErrorCode(0, false)
  /**
    * Thrown when a client sends an RPC request to a node which does not exist.
    */
  object NodeNotFound extends ErrorCode(1, true)
  /**
    * Use this error to indicate that a requested operation is not supported by the current implementation. Helpful for stubbing out APIs during development.
    */
  object NotSupported extends ErrorCode(10, true)
  /**
    * Indicates that the operation definitely cannot be performed at this time--perhaps because the server is in a read-only state, has not yet been initialized, believes its peers to be down, and so on. Do not use this error for indeterminate cases, when the operation may actually have taken place.
    */
  object TemporarilyUnavailable extends ErrorCode(11, true)
  /**
    * The client's request did not conform to the server's expectations, and could not possibly have been processed.
    */
  object MalformedRequest extends ErrorCode(12, true)
  /**
    * Indicates that some kind of general, indefinite error occurred. Use this as a catch-all for errors you can't otherwise categorize, or as a starting point for your error handler: it's safe to return internal-error for every problem by default, then add special cases for more specific errors later.
    */
  object Crash extends ErrorCode(13, false)
  /**
    * Indicates that some kind of general, definite error occurred. Use this as a catch-all for errors you can't otherwise categorize, when you specifically know that the requested operation has not taken place. For instance, you might encounter an indefinite failure during the prepare phase of a transaction: since you haven't started the commit process yet, the transaction can't have taken place. It's therefore safe to return a definite abort to the client.
    */
  object Abort extends ErrorCode(14, true)
  /**
    * The client requested an operation on a key which does not exist (assuming the operation should not automatically create missing keys).
    */
  object KeyDoesNotExist extends ErrorCode(20, true)
  /**
    * The client requested the creation of a key which already exists, and the server will not overwrite it.
    */
  object KeyAlreadyExists extends ErrorCode(21, true)
  /**
    * The requested operation expected some conditions to hold, and those conditions were not met. For instance, a compare-and-set operation might assert that the value of a key is currently 5; if the value is 3, the server would return precondition-failed.
    */
  object PreconditionFailed extends ErrorCode(22, true)
  /**
    * The requested transaction has been aborted because of a conflict with another transaction. Servers need not return this error on every conflict: they may choose to retry automatically instead.
    */
  object TxnConflict extends ErrorCode(30, true)
  /**
    * Custom error code
    *
    * @param code the error code
    */
  case class Custom(override val code: Int) extends ErrorCode(code, false)

You can send an error message to any node id as a reply to another message. Here's an example

Send standard error
case class InMessage() derives JsonCodec

val handler = receive[InMessage](_ =>
  reply(Error(ErrorCode.PreconditionFailed, "some text message")) 
)
Send custom error
case class InMessage() derives JsonCodec

val handler = receive[InMessage](_ =>
  reply(Error(ErrorCode.Custom(1005), "some text message"))
)

Maelstrom services#

Maelstrom starts some services at the beginning of every simulation by default

These are their node ids:

  1. lin-kv
  2. lww-kv
  3. seq-kv
  4. lin-tso

You can read more these services on the maelstrom docs

ZIO-Maelstrom provides LinkKv, LwwKv, SeqKv & LinTso clients to interact with these services. SeqKv, LwwKv & LinKv are all key value stores. They have the same api but different consistency guarantees.

Native KV APIs#

Native apis are provided by the maelstrom services

read

Takes a key and returns the value of the key. If the value does not exist, it returns KeyDoesNotExist error code.

val counterValue: ZIO[SeqKv, AskError, Int] = SeqKv.read("counter", 5.seconds)
write

Takes a key and a value and writes the value against the key. If a value already exists against the key, it is overwritten.

val _: ZIO[LwwKv, AskError, Unit] = LwwKv.write("counter", 1, 5.seconds)
cas

CAS stands for compare-and-swap. It takes a key, a value and an expected value. It writes the value against the key only if the expected value matches the current value of the key. If the value is different, then it returns PreconditionFailed error code. If the key does not exist, it returns KeyDoesNotExist error code. If you set createIfNotExists to true, it will create the key if it does not exist.

val _: ZIO[LinKv, AskError, Unit] =
  LinKv.cas(key = "counter", from = 1, to = 3, createIfNotExists = false, timeout = 5.seconds)

Above example will write 3 to counter only if the current value of counter is 1. If the current value is different, it will return PreconditionFailed error code.

High level KV APIs#

High level apis are built on top of native apis by combining multiple native apis and/or adding additional logic

readOption

Takes a key and returns an Option of the value of the key. If the value does not exist, it returns None. Does not return KeyDoesNotExist error code.

val counterMaybe: ZIO[SeqKv, AskError, Option[Int]] = SeqKv.readOption("counter", 5.seconds)
writeIfNotExists

Takes a key and a value and writes the value against the key only if the key does not exist. If the key already exists, it returns PreconditionFailed error code.

val _: ZIO[LwwKv, AskError, Unit] = LwwKv.writeIfNotExists("counter", 1, 5.seconds)
update

This is a high level api built on top of other apis. It takes a key, a function that takes the current value and returns a new value. It reads the current value of the key, applies the function and writes the new value against the key. If the value has changed in the meantime, it applies the function again and keeps trying until the value does not change. This is useful for implementing atomic operations like incrementing a value.

val increasedValue: ZIO[SeqKv, AskError, Int] = SeqKv.update("counter", 5.seconds) {
  case Some(oldValue) => oldValue + 1
  case None           => 1
}

The timeout value does not apply to entire operation but to each individual read, cas and write operation. So the total time taken by the operation can be more than the timeout value. Retries are only done when the value has changed in the meantime. And other error is returned immediately. This also applies to updateZIO api.

updateZIO

This is a high level api built on top of other apis. It takes a key, a function that takes the current value and returns a ZIO that returns a new value. It reads the current value of the key, applies the ZIO and writes the new value against the key. If the value has changed in the meantime, it applies the function again and keeps trying until the value does not change. This is very similar to update but the function can be a ZIO which can do some async operations. 

def getNewNumber(oldValue: Option[Int]): ZIO[Any, Nothing, Int] = ???

val increasedValueZIO: ZIO[SeqKv, AskError, Int] =
  SeqKv.updateZIO("counter", 5.seconds)(getNewNumber)

Danger

When retries happen, the ZIO is retried as well, so side effects should be avoided in this function.

Important

  • Because all these apis are built on top of ask api, they can return AskError which you may need to handle. According to maelstrom documentation, they can return KeyDoesNotExist or PreconditionFailed error codes.

  • In case of network partition or delay, all of the above apis can return Timeout error code.

  • When incorrect types are used to decode the response, they can return DecodingFailure error code.

Tip

key and value of the key value store can be any type that has a zio.json.JsonCodec instance

TSO APIs#

LinTso is a linearizable timestamp oracle. It has the following api

Linearizable timestamp oracle
val timestamp: ZIO[LinTso, AskError, Int] = LinTso.ts(5.seconds)

Settings#

Below are the settings that can be configured for a node

  1. Log Level

    The default log level is LogLevel.Info. If you want more detailed logs, you can set it to LogLevel.Debug. If you want to disable logs, you can set it to LogLevel.None

  2. Log Format

    Log format can be either Plain or Colored. Default is colored.

  3. Concurrency

    This is the concurrency level for processing messages. Default is 1024. This means 1024 request messages(receive api) + 1024 response messages (ask api) = 2048 messages can be processed in parallel.

Default example
object MainApp extends MaelstromNode {
  val program = ???
}
Customization example
object MainApp extends MaelstromNode {

  override val configure = NodeConfig
    .withConcurrency(100)
    .withLogLevelDebug
    .withPlaintextLog

  val program = ZIO.logDebug("Starting node...")
}

Logging#

You can log at different levels using ZIO's logging APIs - ZIO.logDebug, ZIO.logInfo, etc. All these APIs log to STDERR because STDOUT is used for sending messages. You can configure the log level using settings API. By default, log statements are colored. You can change it to plain using settings API

Logging
object MainApplication extends MaelstromNode {

  override val configure = NodeConfig.withLogLevelDebug

  def program = for
    _ <- ZIO.logDebug("Starting node")
    _ <- ZIO.logInfo("Received message")
    _ <- ZIO.logWarning("Something is wrong")
    _ <- ZIO.logError("Something is really wrong")
  yield ()
}

Above program, when initialized, will output the following:

log-output

Testing#

Using static inline messages:

When developing a solution, you sometimes want to test it without maelstrom. While you can use stdIn to enter the input, you can also hardcode the input messages in the program itself.

Inline Input Messages
object InlineInput extends MaelstromNode {

  case class Ping() derives JsonCodec 

  case class Pong() derives JsonEncoder

  val program = receive[Ping](_ => reply(Pong()))

  override val configure: NodeConfig =
    NodeConfig
      .withStaticInput(
        NodeId("A"),                                                                       
        Set(NodeId("B"), NodeId("C")),                                                     
        InlineMessage(NodeId("B"), Body(MsgName[Ping], Ping(), Some(MessageId(1)), None)), 
        InlineMessage(NodeId("C"), Body(MsgName[Ping], Ping(), Some(MessageId(2)), None))
      )
}

Tip

When debugging an issue, use static input, set log level to debug and set concurrency to 1. This might help you isolate the issue.

Using static input files:

You can configure the runtime to read the input from a file.

With Files
object Main extends MaelstromNode {

  case class Ping() derives JsonDecoder

  case class Pong() derives JsonEncoder

  val program = receive[Ping](_ => reply(Pong()))

  override val configure: NodeConfig =
    NodeConfig
      .withStaticInput(
        NodeId("A"),                          
        Set(NodeId("B"), NodeId("C")),        
        "examples" / "echo" / "fileinput.txt" 
      )
      .withLogLevelDebug
}
fileinput.txt
{"src": "c1", "dest": "n1", "body": { "type" : "ping", "msg_id": 1 }}
sleep 2s
{"src": "c1", "dest": "n1", "body": { "type" : "ping", "msg_id": 2 }}

This will run the entire program with the input from the file. With file input you also get to simulate delay in inputs using sleep statements as shown above.

file-input