S3 log storage writes - CreateIfNotExists guard

[xianwill]

S3 does not have an API that supports the "Create if not exists" semantics that the other storage backends have. This discussion is to work out what a wrapper on top of S3 should look like to provide these semantics when writing delta log entries.

Ultimately, we need the put_obj implementation within the s3 backend to fail with StorageError:AlreadyExists as the fs backend does iff a log entry with the same version has already been written.

DynamoDb is frequently mentioned as a metastore on top of S3 to provide this type of API. delta-rs should provide functionality to:

• create the required DynamoDb schema for guarding log entry writes to S3
• implement a reliable protocol within the s3 backend put_obj method that leverages DynamoDb to guard S3 writes

This discussion is intended to work out the schema and protocol to include "Create if not exists" semantics in the put_obj implementation of the S3 backend.

---

To kick the discussion off with a naive approach:

Consider a DynamoDb schema with only a path attribute. A put_obj implementation may look like:

• Write path to DynamoDb table with a put condition to ensure path does not already exist
• On success - write delta log file to s3
• On fail - continue with optimistic concurrency retry loop

Already, we have some problems. If there is a system failure between DynamoDb guard write and delta log file write for version n, the delta log file for version n will not be written. The delta protocol requires contiguous, monotonically-increasing integers for version so this must be resolved before other log writers can continue the optimistic concurrency loop with any chance of success.

We could implement a distributed lock w/ timeout around the DynamoDb write and S3 log entry file write, but now we have a distributed lock inside of an optimistic concurrency loop which smells pretty funny to me. Perhaps this is tolerable since the goal is to bring the S3 semantics up to speed with other storage backends.

@houqp + @rtyler - your turn now.

[xianwill]

In regards to wrapping the guard and S3 write in a distributed lock, I came across https://crates.io/crates/dynalock recently which may be worth taking a dependency on.

[houqp]

The problem you mentioned exists in our current filesystem backend as well, for example: a writer could crash right after it created the new log entry, but before fully writes out the log content. When this happens, we would end up with a corrupted log entry. The core of the problem here is transaction commit semantic is modeled by creation of the file without taking into account of the file content. Same analysis can be applied to the failure case you mentioned with dynamodb, where putting/reserving a key in dynamodb only indicates "creation" of the file, not a file with fully flushed content.

I think it would be more fitting to change the Create if not exists semantic requirement to Atomic rename/rename if not exists semantic. Take filesystem backend as an example again. What we should do is to write out new transaction log into a temporary path. After the write is fully flushed and persisted, we call a rename system call to rename the temp file to the next available commit version. Atomic rename should guarantee a file already exists error to be thrown if another writer has already claimed this version. This will make sure we never commit corrupted log entries.

Back to S3 + Dynamodb, my original suggestion of leveraging S3's new consistent read capability won't fly as you already pointed out. It looks like to be 100% safe, we would still need to use dyanmodb for listing so we can get that Atomic rename support.

Here is one way to do it:

In our commit loop:

1. Write out log entry to a temp and unique S3 path
2. Atomically put a new key in dynamodb with key set to new transaction version and value set to the above temp S3 path
   • If key already exists, apply all new log entries based on a new dynamodb key scan (this is where we use dynamodb for listing, not s3). We can also store latest version in dynamodb in a dynamodb transaction to avoid the table scan.

To make it backwards compatible with readers using vanilla S3 backends, we can replica logs from dynamodb into S3 in a dedicated single threaded replicator/writer:

1. Go through dyanmodb table to find all unreplicated log entries
2. One by one copy unreplicated log entry into _delta_log folder with name set to the corresponding transaction version conforming to the delta spec
   • Once a log entry is fully copied to S3 with log version as the name, mark this particular log entry in dynamodb as replicated. We can mark it by setting key value to a null or use other metadata fields.
   • Delete the old random S3 log entry

[houqp]

Another way to do it is using the distributed lock design you mentioned to serialize the write:

1. write out log entry to a temp and unique s3 path
2. grab the distributed lock
3. check if next version in s3 log directory already exists (here we are leveraging s3's new consistent read/listing capability)
4. copy temp s3 log entry into log directory with version as the object name
5. release distributed lock

Notice in this design, we only need to use dynamodb for locking. We don't need to use it to implement create if not exists semantic because the lock already guarantees only one writer will be able to reserve s3 path for the next version. Effectively, we are doing Atomic rename by using a distributed lock to safe guard the s3 rename operation. S3 doesn't support native object rename though, so it will be implemented as a list + copy + delete. This will take extra time if the S3 object has a large size, which I don't think is the case for log entries.

This design is a simpler compared to the one that uses dynamodb for listing, at the cost of slower commit time and the need to handle lock expiration edge-case.

[houqp]

There are a lot of of interesting discussions around this topic in delta-io/delta#41 as well, which covered a lot of the points we discussed here.

[xianwill]

Fantastic suggestions all around @houqp. I am all on board for implementing "atomic rename" instead of "create if not exists".

Since the use case I am primarily interested in begins with headless multi-process streaming writers appending to the same table and competing over the next version, I personally think I prefer approach 2 you mention above. I think approach 1 would land me right back into a spot where I would either need a dedicated replication process or another not-yet-mentioned lock to let separate threads in each process compete over which process is doing the replication. Approach 2 does not have this same development overhead as far as i can tell.

[houqp]

Yeah, approach 2 is mostly trading performance and liveness for less complexity. I think putting a lock around S3 copy shouldn't be too bad performance wise because commit log entries are usually small in size. We need to have a solution for the following edge-case though:

1. writer A writes obtains transaction lock, claims log version N
2. writer A starts log entry S3 copy
3. writer A temporarily gets stuck due to bugs or network issue
4. lock times out
5. writer B grabs log and claims the same log version N
6. writer B completes S3 copy, releases lock. at this point, transaction has been committed.
7. writer A back in business, assuming that it still holds the lock, continues to finish up the stale S3 copy.

The system should be able to handle lock timeouts caused by hard crashes without issue. But timeouts caused by this kind of intermittent issues could be tricky to handle.

With regards to approach 1, I would refer us going with a single dedicated replicator to keep it simple since the log copy should be quick and easy. As you already mentioned, distributing it across processes brings us back to the lock liveness issue while approach 1's biggest selling point is lockless.

Lastly, what I mentioned so far are just two of the many possible solutions we can explore :) There might be more elegant solutions out there.

[xianwill]

Re: the edge case - yep. I was thinking about the same thing. My thought is that DynamoDb interactions either need to use DynamoDb transactions, or otherwise include a put-condition that ensures that the held lock has not timed out and handle the failure by resuming the next iteration of the optimistic concurrency loop if the transaction or put-condition fails for lock timeout.

[houqp]

The failure scenario I mentioned in the previous comment is not a hard failure, but rather an unexpected pause. From the first lock holder's point of view, it doesn't know it has been paused for a long time, so after resumed, it will just keep proceeding in the critical section assuming it still owns the expired lock.

https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html has a good demonstration of the brain split problem that I constructed in my previous reply. https://hazelcast.com/blog/long-live-distributed-locks/ is another implementation focused post worth reading.

Usually, this is solved using a fencing token, which is basically a monotonically increasing counter that get issued on every lock operation. However, to effectively use a fencing token, we need all external systems involved within the critical section to support conditional writes, which is not the case for S3. The amazon-dynamodb-lock-client project had a similar discussion around this subject: awslabs/amazon-dynamodb-lock-client#1 (comment).

One thing we can do here is to adopt a 2PC design for the S3 rename. When a writer acquires a lock to perform the rename, it also atomically record this intention in dynamodb by writing out the temp object key. We can think of this as the prepare stage in 2PC. Once the prepare stage is done, we can consider the rename has been completed logically. Once the S3 write has been completed, the writer will perform a final commit in dynamodb to mark the prepared stage as completed.

Now consider the following failure scenarios:

• The first writer crashed after the prepare stage without finishing up the S3 write. The next writer that acquires the expired lock will try to complete the rename by noticing an uncommitted prepare action
• The writer crashed after completed the S3 write, but before committing the rename. The next writer that acquires the expired lock will try to finish up the commit
• The first writer paused after the prepare stage while writing to S3. When the next writer grabs the expired lock, it will try to complete the same S3 rename operation by coping the same temp S3 object to the same prepared transaction log entry. The S3 write is idempotent.
• The first writer paused after completing the S3 write but before committing the prepare stage, it will simply compete with the next writer that acquires the expired lock to commit the same prepared stage. This dynmodb write is also idempotent.

[houqp]

@mosyp @xianwill here is an expanded potential atomic rename design with fencing token.

Before the optimistic commit loop

Write log entries to a temp s3 location (tmp_key = prefix + uuid).

At the beginning of the optimistic commit loop:

Compute the next version from cached delta table version in memory. It is required to compute the version outside of the lock for non-append only writes. For append only writes, we could optimize further to get the latest version from dynanmodb lock.

Atomic rename

1. Acquire dynamodb lock with fencing_token
   • At this point, we have entered the critical section. Keep in mind that because our lock has expiration, it can expire between any atomic operations within the critical section. So the lock itself doesn't provide correctness/safety guarantee.
2. Repair prepared commits if any. These prepared commits are idempotent/static, so it's safe for multiple writers to repair the same commit concurrently
   1. for each prepared commit (tmp_key, delta_version, fencing_token) in dynamodb:
      1. perform s3 write from tmp_key to delta_version
         • if tmp_key is already deleted by another writer, we go straight to mark the prepared commit as committed
      2. perform s3 delete for tmp_key
      3. mark prepared commit as committed in dynamodb
3. Use S3 HEAD API to check if delta_version already exists in S3. If yes, return Object already exists error back to caller.
   • Lock could expire after the S3 version conflict check, that means multiple writers could pass this check with the same version. However, the subsequent dyanmodb conditional put with fencing token will guarantee only one of them goes through.
4. Prepare a commit with the tuple (tmp_key, delta_version, fencing_token). In the dynamodb conditional put, we make sure both fencing_token and delta_version are greater than all existing prepared commit dynamodb items. If conditional PUT failed, return object already exists error back to the caller.
   • Once the dynamodb PUT succeeds, the rename operation has been logically committed in the system and there is no going back. In other words, the prepared rename will eventual become visible to all readers.
   • The current writer can crash or pause from now on and it doesn't matter because other writers will help finish up the prepared commit.
5. "repair" newly prepared commit

[xianwill]

@houqp - I may have spoken unclearly. The case I was referring to when I mentioned

Most likely the S3 backend will have to return an "object exists" error to the caller

is that - assuming a lock was acquired by worker A with intent i1 recorded in lock L1 to commit tmp_key k1 as delta_version v1, and that the lock expired due to worker pause, when worker B tries to prepare a different commit with intent i2, the first operation performed by the S3 backend should be to try to repair failed intents. So rather than acquiring lock L2 to perform intent i2 it should first take over L1 and perform the intent of i1. i1 recorded in L1 will likely have the same version implied by i2 though since worker B will not know about the version, v1, intended by i1 yet (since it was not committed prior to worker B's attempt), so I think this will always result in object already exists for the first iteration of i2.

Am I making sense or have I landed in left field?

[houqp]

Ha, I see. You are right. We are on the same page then. The repair action itself will not cause object already exists error, but subsequent version conflict checks would (step 3 and step 4). When a repair happens, we will always get version conflict in the subsequent version conflict checks.

Here is the scenario play:

• worker A tries to commit v1
• worker A acquires L1
• worker A prepares rename commit k1 -> v1
• worker A pauses and L1 expires
• worker B tries to commit v1
• worker B acquires L2
• worker B repair unfinished commit k1 -> v1 (now v1 s3 object is visible to the world)
• worker B gets object already exists error from S3 Heading v1

I am guessing @mosyp was thinking about the optimization of incrementing the version number for append only writes while the lock is being held?

[xianwill]

I don't want to speak for @mosyp - but when we were talking this morning, I personally needed a "sit and think" period to work out the implications of the lock placement and was not able to do so in real time. Hazarding a guess, I think letting the optimistic concurrency loop handle the version conflict implied by repair will sit fine with him. How about it @mosyp?

[houqp]

If it turned out to be for that optimization, I am also leaning towards doing that as a follow up after we get the reference implementation behavior working end to end first. We likely will only be dealing with single digit dynanmodb calls per second, so there is also the possibility that this will not be a bottleneck for us.

Ideally, the append only version increment optimization should be designed in a way to benefit all storage backends, so it shouldn't depend on a distributed lock.

[mosyp]

@xianwill @houqp Yes, having lock outside of rename was purely an optimisation issue. Also I couldn't figure out repairs then, but your logic of returning version_exists and rely on optimistic concurrency works with me.
Also, I did some testing with lock inside and outside of rename call and couldn't spot much of a difference in #212. Both cases had random execution time in between 9-10 sec each.

So as of now let's keep dynamo locking within s3_rename until we really need otherwise

[mosyp]

@xianwill @houqp wdyt?

Atomic rename for S3 v2.0

We introduce new tmp_path and log_path fields into lock's data object, which represents an attempt of renaming file tmp_path into log_path. These fields are added to the lock only within acquire_lock_with function.

Variables
data1 - data of current lock (if any) that acquired by some worker which in a process of renaming tmp_path1 into log_path1.
data2 - data of this worker that we're trying to rename tmp_path2 into log_path2

The result of acquire_lock_with(tmp_path2, log_path2):

a) The lock is not acquired: try again (still in progress or someone else took it).

b) Acquired released or new lock:

1. Acquire lock with data=tmp_path2/log_path2
2. Call rename(tmp_path2,log_path2)
3. Release lock with data=null

c) Current lock is expired: this is dangerous zone, previous owner is in undefined state - e.g. halting, crashed etc. At this point we have to repair a rename commit (if any) of previous owner:

1. Acquire lock with the same data=tmp_path1/log_path1 (because at this point we only repair failed commit, not creating new).
2. Call rename(tmp_path1,log_path1) e.g. repair failed rename.
3. Release lock with data=null
4. Return with VersionExists so later this worker will have to retry rename(tmp_path2,log_path2) for newer version

[houqp]

Ha, yeah, I think both of you are right, I forgot we have a unique owner token to check when releasing the lock to prevent releasing stale locks. Since we don't need to track progress across multiple transaction steps, a unique lock owner identifier is indeed good enough. Let's keep it simple for now and add fencing token later if a real use-case comes up 👍 .

[mosyp]

That's great, since we're all on the same page, I feel like the most difficult part is behind us and we good to go :)

[houqp]

One thing to watch out for is the (source, dest) pair we store in lock data is not guaranteed to always be valid commit anymore because the conditional put doesn't check for order. The order was enforced in the previous version as a separate conditional PUT step. As a result, in this design we can't always return already_exists error after a commit is being "repaired".

For example:

1. writer A wrote and committed Log version N to table
2. writer B, being behind, acquired the lock with data set to (tmp_b, N)
3. writer B crashed or paused before releasing the lock
4. writer A acquired the lock with data set to (tmp_a, N+1)
5. seeing an existing lock data (tmp_b, N), writer A "repair" the commit and returns already_exists error
6. writer A retries with (tmp_a, N+2), which is not correct

So in case of version going backwards, we need to retry lock with the same version within rename instead of propagate already_exists error back to the optimistic transaction commit loop.

[xianwill]

Whew. Wow. Nice catch. I completely missed that scenario. Agree - very important to make sure we satisfy both monotonic and contiguous properties of each Delta version number.

From PROTOCOL.md:

A table has a single serial history of atomic versions, which are named using contiguous, monotonically-increasing integers.

[mosyp]

Aha, this one #89 (comment), so that why the version_exists was chosen. But having the @houqp example described above, we need to definitely change a current design to avoid returning alread_exists.

I agree with

So in case of version going backwards, we need to retry lock with the same version within rename instead of propagate already_exists error back to the optimistic transaction commit loop

And, even if repair and current rename will resolve into the same version, the latter retry will fail with already_exists and new version will be picked up by optimistic loop instead of always increasing the version