New streaming backend - Githubissues

flashmob commented 5 years ago

Problem

The current 'backend' processes the emails by buffering the entire message in memory, then applying each Processor using the decorator pattern. This approach works OK - most emails are never over 1MB anyway, and the buffers are recycled after processing, keeping them allocated for the next message, (being nice the garbage collector). However, some things can be done more efficiently - for example the message could be compressed while it's being saved on-the-fly. This is the main idea of 'streaming'.

Solution

The go Go way of streaming is to use the io.Writer and io.Reader interfaces. What if we could use the current decorator pattern that we use for the backends, extend that by making the processors implement the io.Writer interface? 🤔

Yes, we can do just that. A little bit of background first: Did you know that the io.Writer way is usually a decorator pattern? When we make an instance of a writer, we usually pass some underlying writer to it, allowing us to wire multiple writers together. Some people call this pattern 'chaining'

Normally, when using io.Writer, if you would like to create a chain, you need to manually wire them with a few lines of code. This solution takes it further, by allowing you to wire the Writers by configuration.

Technical Details

Each Writer is an instance of a StreamDecorator, it's a struct that implements io.Writer. Additionally, the struct contains two callback functions Open and Close, both could be set when the StreamDecorator is being initialized, and called back at the start and end of the stream. The Open callback is also used to pass down the *mail.Envelope which can be used to keep the state for the email currently being processed.

type StreamDecorator struct {
    p     func(StreamProcessor) StreamProcessor
    e     *mail.Envelope
    Close streamCloseWith
    Open  streamOpenWith
}

type streamOpenWith func(e *mail.Envelope) error

type streamCloseWith func() error

in gateway.go there's a new newStreamStack method that instantiates the StreamDecorator structs and wires them up.

A new method was added to the Backend interface

ProcessStream(r io.Reader, e *mail.Envelope) (Result, error)

A new configuration option was also added to the config file: stream_save_process. The value is a string with the names of each StreamDecorator to chain, delimited by a pipe |.

This is how the io.Reader is passed from the DATA command down to the backend. The ProcessStream method calls all the Open methods on our writers, and then begins the streaming of data using io.Copy. At the end of the stream, it calls Close() on our decorators in the other they were wired.

Examples

Perhaps the best way to understand this is to look at some example code.

There are 3 examples of StreamDecorator implementations in the backends dir:

s_header.go - adds a 'delivery header' to the front of the stream
s_compress.go - uses zlib to compress the stream
s_decompress.go - uses zlib to decompress the stream
s_debug - logs the stream

You will notice that each of the files contain a function that looks just like the io.Writer interface, without the Write keyword. I.e StreamProcessWith(func(p []byte) (int, error) This is an anonymous function which is converted to an io.Writerwhen it is returned. Here is the code of s_debug.go

func StreamDebug() *StreamDecorator {
    sd := &StreamDecorator{}
    sd.p =
        func(sp StreamProcessor) StreamProcessor {
            sd.Open = func(e *mail.Envelope) error {
                return nil
            }
            return StreamProcessWith(func(p []byte) (int, error) {
                fmt.Println(string(p))
                Log().WithField("p", string(p)).Info("Debug stream")
                return sp.Write(p)
            })
        }
    return sd
}

The most important detail here is that the sp identifier refers to the next io.Writer in the chain. In other words, sp contains a reference to the underlying writer.

(The sd.Open statement does nothing, it's just there here as an example / to be used as a template.)

In the api_test.go file, there is a test called `TestStreamProcessor'. The writers are chained with the following config setting:

"stream_save_process": "Header|compress|Decompress|debug"

Which means it will call the Write method on the Header first, and then down to each underlying writer in the stream.

Todo:

Configurable stream buffer size & ability to recycle it.
Write more advanced processors (parse headers, streaming version of the maildir processor, sql database, save in chunks & deduplicate on-the-fly)
Fuzz testing
Test in production

flashmob commented 5 years ago

Update: Development ongoing. Got a little side-tracked and developed a streaming parsing library for mime. (it's parsing the headers as the input is read in. In the old version, it had to wait until everything was read in. Just thought it would be nicer if it could also produce a mime tree too, so that took a bit longer, but hopefully the effort was worth it! Re-written it 3 times, lol!). Stream buffer is now configurable, and works with buffers on any size, eg 64 bytes is even OK. Can't wait to try it out in prod, currently only tested on about 1000 sample emails. It would be good to try it out on a few million. Will post the latest source soon...

flashmob commented 5 years ago

Now working on a "chunk saver" - a streaming backend that breaks up the emails in to chunks as they get saved. Chunks are broken according to the following rule: at the end of headers, at the end of mime part boundaries, or if the chunk's size reaches a limit. Chunks can be checked if they are being duplicated & basic data-deduplication can be baked in.

flashmob commented 5 years ago

Just found out that text compresses significantly much better if it's not base64 encoded! (Makes sense... base64 inflates the size by at most ~33%)