Move design draft to separate file; write about GC in internals

doc/book/src/SUMMARY.md

@@ -7,6 +7,8 @@
 - [Cookbook](./cookbook/index.md)
   - [Building from source](./cookbook/from_source.md)
   - [Integration with systemd](./cookbook/systemd.md)
+  - [Gateways](./cookbook/gateways.md)
+  - [Exposing buckets as websites](./cookbook/exposing_websites.md)
   - [Configuring a reverse proxy](./cookbook/reverse_proxy.md)
   - [Production Deployment](./cookbook/real_world.md)
   - [Recovering from failures](./cookbook/recovering.md)
@@ -27,6 +29,7 @@
 - [Design](./design/index.md)
   - [Related Work](./design/related_work.md)
   - [Internals](./design/internals.md)
+  - [Design draft](./design/design_draft.md)
 
 - [Development](./development/index.md)
   - [Setup your environment](./development/devenv.md)

doc/book/src/connect/apps.md

@@ -253,35 +253,7 @@ Make sure you (will) have a corresponding DNS entry for them.
 
 Now we will configure a reverse proxy in front of Garage.
 This is required as we have no other way to serve CORS headers yet.
-For our example, we will use nginx:
+Check the [Configuring a reverse proxy](/cookbook/reverse_proxy.html) section to know how.
-
-```nginx
-server {
-  # In production you should use TLS instead of plain HTTP
-  listen [::]:80;
-
-  server_name peertube-video.web.garage peertube-playlist.web.garage;
-
-  location / {
-    add_header Access-Control-Allow-Origin *;
-    add_header Access-Control-Max-Age 3600;
-    add_header Access-Control-Expose-Headers Content-Length;
-    add_header Access-Control-Allow-Headers Range;
-
-    # We do not forward OPTIONS request to Garage
-    # as it does not know how to interpret them.
-    # Instead, we simply answers 200.
-    if ($request_method !~ ^(GET|HEAD)$ ) {
-      return 200;
-    }
-
-    # If your do not have a Garage instance on the reverse proxy, change the URL here.
-    proxy_pass http://127.0.0.1:3902;
-    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
-    proxy_set_header Host $host;
-  }
-}
-```
 
 Now make sure that your 2 dns entries are pointing to your reverse proxy.
 
@@ -346,13 +318,114 @@ https://docs.joinmastodon.org/admin/config/#cdn
 
 ## Matrix
 
-### synapse-s3-storage-provider
+Matrix is a chat communication protocol. Its main stable server implementation, [Synapse](https://matrix-org.github.io/synapse/latest/), provides a module to store media on a S3 backend. Additionally, a server independent media store supporting S3 has been developped by the community, it has been made possible thanks to how the matrix API has been designed and will work with implementations like Conduit, Dendrite, etc.
 
-https://github.com/matrix-org/synapse-s3-storage-provider
+### synapse-s3-storage-provider (synapse only)
 
-### matrix-media-repo
+Supposing you have a working synapse installation, you can add the module with pip:
 
-https://github.com/turt2live/matrix-media-repo
+```bash
+ pip3 install --user git+https://github.com/matrix-org/synapse-s3-storage-provider.git
+```
+
+Now create a bucket and a key for your matrix instance (note your Key ID and Secret Key somewhere, they will be needed later):
+
+```bash
+garage key new --name matrix-key
+garage bucket create matrix
+garage bucket allow matrix --read --write --key matrix-key
+```
+
+Then you must edit your server configuration (eg. `/etc/matrix-synapse/homeserver.yaml`) and add the `media_storage_providers` root key:
+
+```yaml
+media_storage_providers:
+- module: s3_storage_provider.S3StorageProviderBackend
+  store_local: True    # do we want to store on S3 media created by our users?
+  store_remote: True   # do we want to store on S3 media created
+                       # by users of others servers federated to ours?
+  store_synchronous: True  # do we want to wait that the file has been written before returning?
+  config:
+    bucket: matrix       # the name of our bucket, we chose matrix earlier
+    region_name: garage  # only "garage" is supported for the region field
+    endpoint_url: http://localhost:3900 # the path to the S3 endpoint
+    access_key_id: "GKxxx" # your Key ID
+    secret_access_key: "xxxx" # your Secret Key
+```
+
+Note that uploaded media will also be stored locally and this behavior can not be deactivated, it is even required for
+some operations like resizing images.
+In fact, your local filesysem is considered as a cache but without any automated way to garbage collect it.
+
+We can build our garbage collector with `s3_media_upload`, a tool provided with the module.
+If you installed the module with the command provided before, you should be able to bring it in your path:
+
+```
+PATH=$HOME/.local/bin/:$PATH
+command -v s3_media_upload
+```
+
+Now we can write a simple script (eg `~/.local/bin/matrix-cache-gc`):
+
+```bash
+#!/bin/bash
+
+## CONFIGURATION ##
+AWS_ACCESS_KEY_ID=GKxxx
+AWS_SECRET_ACCESS_KEY=xxxx
+S3_ENDPOINT=http://localhost:3900
+S3_BUCKET=matrix
+MEDIA_STORE=/var/lib/matrix-synapse/media
+PG_USER=matrix
+PG_PASS=xxxx
+PG_DB=synapse
+PG_HOST=localhost
+PG_PORT=5432
+
+## CODE ##
+PATH=$HOME/.local/bin/:$PATH
+cat > database.yaml <<EOF 
+user: $PG_USER
+password: $PG_PASS
+database: $PG_DB
+host: $PG_HOST
+port: $PG_PORT
+EOF
+
+s3_media_upload update-db 1d
+s3_media_upload --no-progress check-deleted $MEDIA_STORE
+s3_media_upload --no-progress upload $MEDIA_STORE $S3_BUCKET --delete --endpoint-url $S3_ENDPOINT
+```
+
+This script will list all the medias that were not accessed in the 24 hours according to your database.
+It will check if, in this list, the file still exists in the local media store.
+For files that are still in the cache, it will upload them to S3 if they are not already present (in case of a crash or an initial synchronisation).
+Finally, the script will delete these files from the cache.
+
+Make this script executable and check that it works:
+
+```bash
+chmod +x $HOME/.local/bin/matrix-cache-gc
+matrix-cache-gc
+```
+
+Add it to your crontab. Open the editor with:
+
+```bash
+crontab -e
+```
+
+And add a new line. For example, to run it every 10 minutes:
+
+```cron
+*/10 * * * * $HOME/.local/bin/matrix-cache-gc
+```
+
+*External link:* [Github > matrix-org/synapse-s3-storage-provider](https://github.com/matrix-org/synapse-s3-storage-provider)
+
+### matrix-media-repo (server independent)
+
+*External link:* [matrix-media-repo Documentation > S3](https://docs.t2bot.io/matrix-media-repo/configuration/s3-datastore.html)
 
 ## Pixelfed
 

doc/book/src/connect/websites.md

@@ -1 +1,77 @@
 # Websites (Hugo, Jekyll, Publii...)
+
+Garage is also suitable to host static websites.
+While they can be deployed with traditional CLI tools, some static website generators have integrated options to ease your workflow.
+
+## Hugo
+
+Add to your `config.toml` the following section:
+
+```toml
+[[deployment.targets]]
+ URL = "s3://<bucket>?endpoint=<endpoint>&disableSSL=<bool>&s3ForcePathStyle=true&region=garage"
+```
+
+For example:
+
+```toml
+[[deployment.targets]]
+ URL = "s3://my-blog?endpoint=localhost:9000&disableSSL=true&s3ForcePathStyle=true&region=garage"
+```
+
+Then inform hugo of your credentials:
+
+```bash
+export AWS_ACCESS_KEY_ID=GKxxx
+export AWS_SECRET_ACCESS_KEY=xxx
+```
+
+And finally deploy your website:
+
+```bsh
+hugo deploy
+```
+
+*External links:*
+  - [gocloud.dev > aws > Supported URL parameters](https://pkg.go.dev/gocloud.dev/aws?utm_source=godoc#ConfigFromURLParams)
+  - [Hugo Documentation > hugo deploy](https://gohugo.io/hosting-and-deployment/hugo-deploy/)
+
+## Publii
+
+It would require a patch either on Garage or on Publii to make both systems work.
+
+Currently, the proposed workaround is to deploy your website manually:
+  - On the left menu, click on Server, choose Manual Deployment (the logo looks like a compressed file)
+  - Set your website URL, keep Output type as "Non-compressed catalog"
+  - Click on Save changes
+  - Click on Sync your website (bottom left of the app)
+  - On the new page, click again on Sync your website
+  - Click on Get website files
+  - You need to synchronize the output folder you see in your file explorer, we will use minio client.
+
+Be sure that you [configured minio client](cli.html#minio-client-recommended).
+
+Then copy this output folder
+
+```bash
+mc mirror --overwrite output garage/my-site
+```
+
+## Generic (eg. Jekyll)
+
+Some tools do not support sending to a S3 backend but output a compiled folder on your system.
+We can then use any CLI tool to upload this content to our S3 target.
+
+First, start by [configuring minio client](cli.html#minio-client-recommended).
+
+Then build your website:
+
+```bash
+jekyll build
+```
+
+And copy jekyll's output folder on S3:
+
+```bash
+mc mirror --overwrite _site garage/my-site
+```

doc/book/src/cookbook/exposing_websites.md

@@ -0,0 +1,48 @@
+# Exposing buckets as websites
+
+You can expose your bucket as a website with this simple command:
+
+```bash
+garage bucket website --allow my-website
+```
+
+Now it will be **publicly** exposed on the web endpoint (by default listening on port 3902).
+
+Our website serving logic is as follow:
+  - Supports only static websites (no support for PHP or other languages)
+  - Does not support directory listing
+  - The index is defined in your `garage.toml`. ([ref](/reference_manual/configuration.html#index))
+
+Now we need to infer the URL of your website through your bucket name.
+Let assume:
+  - we set `root_domain = ".web.example.com"` in `garage.toml` ([ref](/reference_manual/configuration.html#root_domain))
+  - our bucket name is `garagehq.deuxfleurs.fr`.
+
+Our bucket will be served if the Host field matches one of these 2 values (the port is ignored):
+
+  - `garagehq.deuxfleurs.fr.web.example.com`: you can dedicate a subdomain to your users (here `web.example.com`).
+
+  - `garagehq.deuxfleurs.fr`: your users can bring their own domain name, they just need to point them to your Garage cluster.
+
+You can try this logic locally, without configuring any DNS, thanks to `curl`:
+
+```bash
+# prepare your test
+echo hello world > /tmp/index.html
+mc cp /tmp/index.html garage/garagehq.deuxfleurs.fr
+
+curl -H 'Host: garagehq.deuxfleurs.fr' http://localhost:3902
+# should print "hello world"
+
+curl -H 'Host: garagehq.deuxfleurs.fr.web.example.com' http://localhost:3902
+# should also print "hello world"
+```
+
+Now that you understand how website logic works on Garage, you can:
+
+ - make the website endpoint listens on port 80 (instead of 3902)
+ - use iptables to redirect the port 80 to the port 3902:  
+   `iptables -t nat -A PREROUTING -p tcp -dport 80 -j REDIRECT -to-port 3902`
+ - or configure a [reverse proxy](reverse_proxy.html) in front of Garage to add TLS (HTTPS), CORS support, etc.
+
+You can also take a look at [Website Integration](/connect/websites.html) to see how you can add Garage to your workflow.

doc/book/src/cookbook/gateways.md

@@ -0,0 +1,31 @@
+# Gateways
+
+Gateways allow you to expose Garage endpoints (S3 API and websites) without storing data on the node.
+
+## Benefits
+
+You can configure Garage as a gateway on all nodes that will consume your S3 API, it will provide you the following benefits:
+
+  - **It removes 1 or 2 network RTT** Instead of (querying your reverse proxy then) querying a random node of the cluster that will forward your request to the nodes effectively storing the data, your local gateway will directly knows which node to query. 
+
+  - **It ease server management** Instead of tracking in your reverse proxy and DNS what are the current Garage nodes, your gateway being part of the cluster keeps this information for you. In your software, you will always specify `http://localhost:3900`.
+
+  - **It simplifies security** Instead of having to maintain and renew a TLS certificate, you leverage the Secret Handshake protocol we use for our cluster. The S3 API protocol will be in plain text but limited to your local machine.
+
+## Limitations
+
+Currently it will not work with minio client. Follow issue [#64](https://git.deuxfleurs.fr/Deuxfleurs/garage/issues/64) for more information.
+
+## Spawn a Gateway
+
+The instructions are similar to a regular node, the only option that is different is while configuring the node, you must set the `--gateway` parameter:
+
+```bash
+garage node configure --gateway --tag gw1 xxxx
+```
+
+Then use `http://localhost:3900` when a S3 endpoint is required:
+
+```bash
+aws --endpoint-url http://127.0.0.1:3900 s3 ls
+```

doc/book/src/cookbook/real_world.md

@@ -78,6 +78,8 @@ rpc_bind_addr = "[::]:3901"
 rpc_public_addr = "<this node's public IP>:3901"
 rpc_secret = "<RPC secret>"
 
+bootstrap_peers = []
+
 [s3_api]
 s3_region = "garage"
 api_bind_addr = "[::]:3900"

doc/book/src/cookbook/reverse_proxy.md

@@ -1 +1,165 @@
 # Configuring a reverse proxy
+
+The main reason to add a reverse proxy in front of Garage is to provide TLS to your users.
+
+In production you will likely need your certificates signed by a certificate authority.
+The most automated way is to use a provider supporting the [ACME protocol](https://datatracker.ietf.org/doc/html/rfc8555) 
+such as [Let's Encrypt](https://letsencrypt.org/), [ZeroSSL](https://zerossl.com/) or [Buypass Go SSL](https://www.buypass.com/ssl/products/acme).
+
+If you are only testing Garage, you can generate a self-signed certificate to follow the documentation:
+
+```bash
+openssl req \
+    -new \
+    -x509 \
+    -keyout /tmp/garage.key \
+    -out /tmp/garage.crt \
+    -nodes  \
+    -subj "/C=XX/ST=XX/L=XX/O=XX/OU=XX/CN=localhost/emailAddress=X@X.XX" \
+    -addext "subjectAltName = DNS:localhost, IP:127.0.0.1"
+
+cat /tmp/garage.key /tmp/garage.crt > /tmp/garage.pem
+```
+
+Be careful as you will need to allow self signed certificates in your client.
+For example, with minio, you must add the `--insecure` flag.
+An example:
+
+```bash
+mc ls --insecure garage/
+```
+
+## socat (only for testing purposes)
+
+If you want to test Garage with a TLS frontend, socat can do it for you in a single command:
+
+```bash
+socat \
+"openssl-listen:443,\
+reuseaddr,\
+fork,\
+verify=0,\
+cert=/tmp/garage.pem" \
+tcp4-connect:localhost:3900
+```
+
+## Nginx
+
+Nginx is a well-known reverse proxy suitable for production.
+We do the configuration in 3 steps: first we define the upstream blocks ("the backends")
+then we define the server blocks ("the frontends") for the S3 endpoint and finally for the web endpoint.
+
+The following configuration blocks can be all put in the same `/etc/nginx/sites-available/garage.conf`.
+To make your configuration active, run `ln -s /etc/nginx/sites-available/garage.conf /etc/nginx/sites-enabled/`.
+If you directly put the instructions in the root `nginx.conf`, keep in mind that these configurations must be enclosed inside a `http { }` block.
+
+And do not forget to reload nginx with `systemctl reload nginx` or `nginx -s reload`.
+
+### Defining backends
+
+First, we need to tell to nginx how to access our Garage cluster.
+Because we have multiple nodes, we want to leverage all of them by spreading the load.
+
+In nginx, we can do that with the upstream directive.
+Because we have 2 endpoints: one for the S3 API and one to serve websites,
+we create 2 backends named respectively `s3_backend` and `web_backend`.
+
+A documented example for the `s3_backend` assuming you chose port 3900:
+
+```nginx
+upstream s3_backend {
+  # if you have a garage instance locally
+  server 127.0.0.1:3900;
+  # you can also put your other instances
+  server 192.168.1.3:3900;
+  # domain names also work
+  server garage1.example.com:3900;
+  # you can assign weights if you have some servers 
+  # that are more powerful than others
+  server garage2.example.com:3900 weight=2;
+}
+```
+
+A similar example for the `web_backend` assuming you chose port 3902:
+
+```nginx
+upstream web_backend {
+  server 127.0.0.1:3902;
+  server 192.168.1.3:3902;
+  server garage1.example.com:3902;
+  server garage2.example.com:3902 weight=2;
+}
+```
+
+### Exposing the S3 API
+
+The configuration section for the S3 API is simple as we only support path-access style yet.
+We simply configure the TLS parameters and forward all the requests to the backend:
+
+```nginx
+server {
+  listen [::]:443 http2 ssl;
+  ssl_certificate     /tmp/garage.crt;
+  ssl_certificate_key /tmp/garage.key;
+
+  # should be the endpoint you want
+  # aws uses s3.amazonaws.com for example
+  server_name garage.example.com;
+
+  location / {
+    proxy_pass http://s3_backend;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_set_header Host $host;
+  }
+}
+
+```
+
+### Exposing the web endpoint
+
+The web endpoint is a bit more complicated to configure as it listens on many different `Host` fields.
+To better understand the logic involved, you can refer to the [Exposing buckets as websites](/cookbook/exposing_websites.html) section.
+Also, for some applications, you may need to serve CORS headers: Garage can not serve them directly but we show how we can use nginx to serve them.
+You can use the following example as your starting point:
+
+```nginx
+server {
+  listen [::]:443 http2 ssl;
+  ssl_certificate     /tmp/garage.crt;
+  ssl_certificate_key /tmp/garage.key;
+
+  # We list all the Hosts fields that can access our buckets
+  server_name *.web.garage
+              example.com
+              my-site.tld
+              ;
+
+  location / {
+    # Add these headers only if you want to allow CORS requests
+    # For production use, more specific rules would be better for your security
+    add_header Access-Control-Allow-Origin *;
+    add_header Access-Control-Max-Age 3600;
+    add_header Access-Control-Expose-Headers Content-Length;
+    add_header Access-Control-Allow-Headers Range;
+
+    # We do not forward OPTIONS requests to Garage
+    # as it does not support them but they are needed for CORS.
+    if ($request_method = OPTIONS) {
+      return 200;
+    }
+
+    proxy_pass http://web_backend;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_set_header Host $host;
+  }
+}
+```
+
+
+## Apache httpd
+
+@TODO
+
+## Traefik
+
+@TODO

doc/book/src/cookbook/systemd.md

@@ -1,9 +1,14 @@
-# Starting Garage with systemd instead of Docker
+# Starting Garage with systemd
+
+We make some assumptions for this systemd deployment. 
+
+  - Your garage binary is located at `/usr/local/bin/garage`.
+
+  - Your configuration file is located at `/etc/garage.toml`.
+
+  - Your `garage.toml` must be set with  `metadata_dir=/var/lib/garage/meta` and `data_dir=/var/lib/garage/data`. This is mandatory to use `systemd` hardening feature [Dynamic User](https://0pointer.net/blog/dynamic-users-with-systemd.html). Note that in your host filesystem, Garage data will be held in `/var/lib/private/garage`.
 
-NOTE: This guide is incomplete. Typicall you would also want to create a separate
-Unix user to run Garage.
 
-Make sure you have the Garage binary installed on your system (see [quick start](../quick_start/index.md)), e.g. at `/usr/local/bin/garage`.
 
 Create a file named `/etc/systemd/system/garage.service`:
 
@@ -15,12 +20,18 @@ Wants=network-online.target
 
 [Service]
 Environment='RUST_LOG=garage=info' 'RUST_BACKTRACE=1'
-ExecStart=/usr/local/bin/garage server -c /etc/garage/garage.toml
+ExecStart=/usr/local/bin/garage server
+StateDirectory=garage
+DynamicUser=true
+ProtectHome=true
+NoNewPrivileges=true
 
 [Install]
 WantedBy=multi-user.target
 ```
 
+*A note on hardening: garage will be run as a non privileged user, its user id is dynamically allocated by systemd. It cannot access (read or write) home folders (/home, /root and /run/user), the rest of the filesystem can only be read but not written, only the path seen as /var/lib/garage is writable as seen by the service (mapped to /var/lib/private/garage on your host). Additionnaly, the process can not gain new privileges over time.*
+
 To start the service then automatically enable it at boot:
 
 ```bash

doc/book/src/design/design_draft.md

@@ -0,0 +1,162 @@
+# Design draft
+
+**WARNING: this documentation is a design draft which was written before Garage's actual implementation.
+The general principle are similar, but details have not been updated.**
+
+
+#### Modules
+
+- `membership/`: configuration, membership management (gossip of node's presence and status), ring generation --> what about Serf (used by Consul/Nomad) : https://www.serf.io/? Seems a huge library with many features so maybe overkill/hard to integrate
+- `metadata/`: metadata management
+- `blocks/`: block management, writing, GC and rebalancing
+- `internal/`: server to server communication (HTTP server and client that reuses connections, TLS if we want, etc)
+- `api/`: S3 API
+- `web/`: web management interface
+
+#### Metadata tables
+
+**Objects:**
+
+- *Hash key:* Bucket name (string)
+- *Sort key:* Object key (string)
+- *Sort key:* Version timestamp (int)
+- *Sort key:* Version UUID (string)
+- Complete: bool
+- Inline: bool, true for objects < threshold (say 1024)
+- Object size (int)
+- Mime type (string)
+- Data for inlined objects (blob)
+- Hash of first block otherwise (string)
+
+*Having only a hash key on the bucket name will lead to storing all file entries of this table for a specific bucket on a single node. At the same time, it is the only way I see to rapidly being able to list all bucket entries...*
+
+**Blocks:**
+
+- *Hash key:* Version UUID (string)
+- *Sort key:* Offset of block in total file (int)
+- Hash of data block (string)
+
+A version is defined by the existence of at least one entry in the blocks table for a certain version UUID.
+We must keep the following invariant: if a version exists in the blocks table, it has to be referenced in the objects table.
+We explicitly manage concurrent versions of an object: the version timestamp and version UUID columns are index columns, thus we may have several concurrent versions of an object.
+Important: before deleting an older version from the objects table, we must make sure that we did a successfull delete of the blocks of that version from the blocks table.
+
+Thus, the workflow for reading an object is as follows:
+
+1. Check permissions (LDAP)
+2. Read entry in object table. If data is inline, we have its data, stop here.
+   -> if several versions, take newest one and launch deletion of old ones in background
+3. Read first block from cluster. If size <= 1 block, stop here.
+4. Simultaneously with previous step, if size > 1 block: query the Blocks table for the IDs of the next blocks
+5. Read subsequent blocks from cluster
+
+Workflow for PUT:
+
+1. Check write permission (LDAP)
+2. Select a new version UUID
+3. Write a preliminary entry for the new version in the objects table with complete = false
+4. Send blocks to cluster and write entries in the blocks table
+5. Update the version with complete = true and all of the accurate information (size, etc)
+6. Return success to the user
+7. Launch a background job to check and delete older versions
+
+Workflow for DELETE:
+
+1. Check write permission (LDAP)
+2. Get current version (or versions) in object table
+3. Do the deletion of those versions NOT IN A BACKGROUND JOB THIS TIME
+4. Return succes to the user if we were able to delete blocks from the blocks table and entries from the object table
+
+To delete a version:
+
+1. List the blocks from Cassandra
+2. For each block, delete it from cluster. Don't care if some deletions fail, we can do GC.
+3. Delete all of the blocks from the blocks table
+4. Finally, delete the version from the objects table
+
+Known issue: if someone is reading from a version that we want to delete and the object is big, the read might be interrupted. I think it is ok to leave it like this, we just cut the connection if data disappears during a read.
+
+("Soit P un problème, on s'en fout est une solution à ce problème")
+
+#### Block storage on disk
+
+**Blocks themselves:**
+
+- file path = /blobs/(first 3 hex digits of hash)/(rest of hash)
+
+**Reverse index for GC & other block-level metadata:**
+
+- file path = /meta/(first 3 hex digits of hash)/(rest of hash)
+- map block hash -> set of version UUIDs where it is referenced
+
+Usefull metadata:
+
+- list of versions that reference this block in the Casandra table, so that we can do GC by checking in Cassandra that the lines still exist
+- list of other nodes that we know have acknowledged a write of this block, usefull in the rebalancing algorithm
+
+Write strategy: have a single thread that does all write IO so that it is serialized (or have several threads that manage independent parts of the hash space). When writing a blob, write it to a temporary file, close, then rename so that a concurrent read gets a consistent result (either not found or found with whole content).
+
+Read strategy: the only read operation is get(hash) that returns either the data or not found (can do a corruption check as well and return corrupted state if it is the case). Can be done concurrently with writes.
+
+**Internal API:**
+
+- get(block hash) -> ok+data/not found/corrupted
+- put(block hash & data, version uuid + offset) -> ok/error
+- put with no data(block hash, version uuid + offset) -> ok/not found plz send data/error
+- delete(block hash, version uuid + offset) -> ok/error
+
+GC: when last ref is deleted, delete block.
+Long GC procedure: check in Cassandra that version UUIDs still exist and references this block.
+
+Rebalancing: takes as argument the list of newly added nodes.
+
+- List all blocks that we have. For each block:
+- If it hits a newly introduced node, send it to them.
+  Use put with no data first to check if it has to be sent to them already or not.
+  Use a random listing order to avoid race conditions (they do no harm but we might have two nodes sending the same thing at the same time thus wasting time).
+- If it doesn't hit us anymore, delete it and its reference list.
+
+Only one balancing can be running at a same time. It can be restarted at the beginning with new parameters.
+
+#### Membership management
+
+Two sets of nodes:
+
+- set of nodes from which a ping was recently received, with status: number of stored blocks, request counters, error counters, GC%, rebalancing%
+  (eviction from this set after say 30 seconds without ping)
+- set of nodes that are part of the system, explicitly modified by the operator using the web UI (persisted to disk),
+  is a CRDT using a version number for the value of the whole set
+
+Thus, three states for nodes:
+
+- healthy: in both sets
+- missing: not pingable but part of desired cluster
+- unused/draining: currently present but not part of the desired cluster, empty = if contains nothing, draining = if still contains some blocks
+
+Membership messages between nodes:
+
+- ping with current state + hash of current membership info -> reply with same info
+- send&get back membership info (the ids of nodes that are in the two sets): used when no local membership change in a long time and membership info hash discrepancy detected with first message (passive membership fixing with full CRDT gossip)
+- inform of newly pingable node(s) -> no result, when receive new info repeat to all (reliable broadcast)
+- inform of operator membership change -> no result, when receive new info repeat to all (reliable broadcast)
+
+Ring: generated from the desired set of nodes, however when doing read/writes on the ring, skip nodes that are known to be not pingable.
+The tokens are generated in a deterministic fashion from node IDs (hash of node id + token number from 1 to K).
+Number K of tokens per node: decided by the operator & stored in the operator's list of nodes CRDT. Default value proposal: with node status information also broadcast disk total size and free space, and propose a default number of tokens equal to 80%Free space / 10Gb. (this is all user interface)
+
+
+#### Constants
+
+- Block size: around 1MB ? --> Exoscale use 16MB chunks
+- Number of tokens in the hash ring: one every 10Gb of allocated storage
+- Threshold for storing data directly in Cassandra objects table: 1kb bytes (maybe up to 4kb?)
+- Ping timeout (time after which a node is registered as unresponsive/missing): 30 seconds
+- Ping interval: 10 seconds
+- ??
+
+#### Links
+
+- CDC: <https://www.usenix.org/system/files/conference/atc16/atc16-paper-xia.pdf>
+- Erasure coding: <http://web.eecs.utk.edu/~jplank/plank/papers/CS-08-627.html>
+- [Openstack Storage Concepts](https://docs.openstack.org/arch-design/design-storage/design-storage-concepts.html)
+- [RADOS](https://ceph.com/wp-content/uploads/2016/08/weil-rados-pdsw07.pdf)

doc/book/src/design/internals.md

@@ -1,158 +1,95 @@
-**WARNING: this documentation is more a "design draft", which was written before Garage's actual implementation. The general principle is similar but details have not yet been updated.**
+# Internals
 
-#### Modules
+## Overview
 
-- `membership/`: configuration, membership management (gossip of node's presence and status), ring generation --> what about Serf (used by Consul/Nomad) : https://www.serf.io/? Seems a huge library with many features so maybe overkill/hard to integrate
+TODO: write this section
-- `metadata/`: metadata management
-- `blocks/`: block management, writing, GC and rebalancing
-- `internal/`: server to server communication (HTTP server and client that reuses connections, TLS if we want, etc)
-- `api/`: S3 API
-- `web/`: web management interface
 
-#### Metadata tables
+- The Dynamo ring
 
-**Objects:**
+- CRDTs
 
-- *Hash key:* Bucket name (string)
+- Consistency model of Garage tables
-- *Sort key:* Object key (string)
-- *Sort key:* Version timestamp (int)
-- *Sort key:* Version UUID (string)
-- Complete: bool
-- Inline: bool, true for objects < threshold (say 1024)
-- Object size (int)
-- Mime type (string)
-- Data for inlined objects (blob)
-- Hash of first block otherwise (string)
 
-*Having only a hash key on the bucket name will lead to storing all file entries of this table for a specific bucket on a single node. At the same time, it is the only way I see to rapidly being able to list all bucket entries...*
+See this presentation (in French) for some first information:
-
+<https://git.deuxfleurs.fr/Deuxfleurs/garage/src/branch/main/doc/talks/2020-12-02_wide-team/talk.pdf>
-**Blocks:**
-
-- *Hash key:* Version UUID (string)
-- *Sort key:* Offset of block in total file (int)
-- Hash of data block (string)
-
-A version is defined by the existence of at least one entry in the blocks table for a certain version UUID.
-We must keep the following invariant: if a version exists in the blocks table, it has to be referenced in the objects table.
-We explicitly manage concurrent versions of an object: the version timestamp and version UUID columns are index columns, thus we may have several concurrent versions of an object.
-Important: before deleting an older version from the objects table, we must make sure that we did a successfull delete of the blocks of that version from the blocks table.
-
-Thus, the workflow for reading an object is as follows:
-
-1. Check permissions (LDAP)
-2. Read entry in object table. If data is inline, we have its data, stop here.
-   -> if several versions, take newest one and launch deletion of old ones in background
-3. Read first block from cluster. If size <= 1 block, stop here.
-4. Simultaneously with previous step, if size > 1 block: query the Blocks table for the IDs of the next blocks
-5. Read subsequent blocks from cluster
-
-Workflow for PUT:
-
-1. Check write permission (LDAP)
-2. Select a new version UUID
-3. Write a preliminary entry for the new version in the objects table with complete = false
-4. Send blocks to cluster and write entries in the blocks table
-5. Update the version with complete = true and all of the accurate information (size, etc)
-6. Return success to the user
-7. Launch a background job to check and delete older versions
-
-Workflow for DELETE:
-
-1. Check write permission (LDAP)
-2. Get current version (or versions) in object table
-3. Do the deletion of those versions NOT IN A BACKGROUND JOB THIS TIME
-4. Return succes to the user if we were able to delete blocks from the blocks table and entries from the object table
-
-To delete a version:
-
-1. List the blocks from Cassandra
-2. For each block, delete it from cluster. Don't care if some deletions fail, we can do GC.
-3. Delete all of the blocks from the blocks table
-4. Finally, delete the version from the objects table
-
-Known issue: if someone is reading from a version that we want to delete and the object is big, the read might be interrupted. I think it is ok to leave it like this, we just cut the connection if data disappears during a read.
-
-("Soit P un problème, on s'en fout est une solution à ce problème")
-
-#### Block storage on disk
-
-**Blocks themselves:**
-
-- file path = /blobs/(first 3 hex digits of hash)/(rest of hash)
-
-**Reverse index for GC & other block-level metadata:**
-
-- file path = /meta/(first 3 hex digits of hash)/(rest of hash)
-- map block hash -> set of version UUIDs where it is referenced
-
-Usefull metadata:
-
-- list of versions that reference this block in the Casandra table, so that we can do GC by checking in Cassandra that the lines still exist
-- list of other nodes that we know have acknowledged a write of this block, usefull in the rebalancing algorithm
-
-Write strategy: have a single thread that does all write IO so that it is serialized (or have several threads that manage independent parts of the hash space). When writing a blob, write it to a temporary file, close, then rename so that a concurrent read gets a consistent result (either not found or found with whole content).
-
-Read strategy: the only read operation is get(hash) that returns either the data or not found (can do a corruption check as well and return corrupted state if it is the case). Can be done concurrently with writes.
-
-**Internal API:**
-
-- get(block hash) -> ok+data/not found/corrupted
-- put(block hash & data, version uuid + offset) -> ok/error
-- put with no data(block hash, version uuid + offset) -> ok/not found plz send data/error
-- delete(block hash, version uuid + offset) -> ok/error
-
-GC: when last ref is deleted, delete block.
-Long GC procedure: check in Cassandra that version UUIDs still exist and references this block.
-
-Rebalancing: takes as argument the list of newly added nodes.
-
-- List all blocks that we have. For each block:
-- If it hits a newly introduced node, send it to them.
-  Use put with no data first to check if it has to be sent to them already or not.
-  Use a random listing order to avoid race conditions (they do no harm but we might have two nodes sending the same thing at the same time thus wasting time).
-- If it doesn't hit us anymore, delete it and its reference list.
-
-Only one balancing can be running at a same time. It can be restarted at the beginning with new parameters.
-
-#### Membership management
-
-Two sets of nodes:
-
-- set of nodes from which a ping was recently received, with status: number of stored blocks, request counters, error counters, GC%, rebalancing%
-  (eviction from this set after say 30 seconds without ping)
-- set of nodes that are part of the system, explicitly modified by the operator using the web UI (persisted to disk),
-  is a CRDT using a version number for the value of the whole set
-
-Thus, three states for nodes:
-
-- healthy: in both sets
-- missing: not pingable but part of desired cluster
-- unused/draining: currently present but not part of the desired cluster, empty = if contains nothing, draining = if still contains some blocks
-
-Membership messages between nodes:
-
-- ping with current state + hash of current membership info -> reply with same info
-- send&get back membership info (the ids of nodes that are in the two sets): used when no local membership change in a long time and membership info hash discrepancy detected with first message (passive membership fixing with full CRDT gossip)
-- inform of newly pingable node(s) -> no result, when receive new info repeat to all (reliable broadcast)
-- inform of operator membership change -> no result, when receive new info repeat to all (reliable broadcast)
-
-Ring: generated from the desired set of nodes, however when doing read/writes on the ring, skip nodes that are known to be not pingable.
-The tokens are generated in a deterministic fashion from node IDs (hash of node id + token number from 1 to K).
-Number K of tokens per node: decided by the operator & stored in the operator's list of nodes CRDT. Default value proposal: with node status information also broadcast disk total size and free space, and propose a default number of tokens equal to 80%Free space / 10Gb. (this is all user interface)
 
 
-#### Constants
+## Garbage collection
 
-- Block size: around 1MB ? --> Exoscale use 16MB chunks
+A faulty garbage collection procedure has been the cause of
-- Number of tokens in the hash ring: one every 10Gb of allocated storage
+[critical bug #39](https://git.deuxfleurs.fr/Deuxfleurs/garage/issues/39).
-- Threshold for storing data directly in Cassandra objects table: 1kb bytes (maybe up to 4kb?)
+This precise bug was fixed in the code, however there are potentially more
-- Ping timeout (time after which a node is registered as unresponsive/missing): 30 seconds
+general issues with the garbage collector being too eager and deleting things
-- Ping interval: 10 seconds
+too early. This has been the subject of
-- ??
+[PR #135](https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/135).
+This section summarizes the discussions on this topic.
 
-#### Links
+Rationale: we want to ensure Garage's safety by making sure things don't get
+deleted from disk if they are still needed. Two aspects are involved in this.
+
+### 1. Garbage collection of table entries (in `meta/` directory)
+
+The `Entry` trait used for table entries (defined in `tables/schema.rs`)
+defines a function `is_tombstone()` that returns `true` if that entry
+represents an entry that is deleted in the table. CRDT semantics by default
+keep all tombstones, because they are necessary for reconciliation: if node A
+has a tombstone that supersedes a value `x`, and node B has value `x`, A has to
+keep the tombstone in memory so that the value `x` can be properly deleted at
+node `B`. Otherwise, due to the CRDT reconciliation rule, the value `x` from B
+would flow back to A and a deleted item would reappear in the system.
+
+Here, we have some control on the nodes involved in storing Garage data.
+Therefore we have a garbage collector that is able to delete tombstones UNDER
+CERTAIN CONDITIONS. This garbage collector is implemented in `table/gc.rs`. To
+delete a tombstone, the following condition has to be met:
+
+- All nodes responsible for storing this entry are aware of the existence of
+  the tombstone, i.e. they cannot hold another version of the entry that is
+  superseeded by the tombstone. This ensures that deleting the tombstone is
+  safe and that no deleted value will come back in the system.
+
+Garage makes use of Sled's atomic operations (such as compare-and-swap and
+transactions) to ensure that only tombstones that have been correctly
+propagated to other nodes are ever deleted from the local entry tree.
+
+This GC is safe in the following sense: no non-tombstone data is ever deleted
+from Garage tables.
+
+**However**, there is an issue with the way this interacts with data
+rebalancing in the case when a partition is moving between nodes. If a node has
+some data of a partition for which it is not responsible, it has to offload it.
+However that offload process takes some time. In that interval, the GC does not
+check with that node if it has the tombstone before deleting the tombstone, so
+perhaps it doesn't have it and when the offload finally happens, old data comes
+back in the system.
+
+**PR 135 mostly fixes this** by implementing a 24-hour delay before anything is
+garbage collected in a table. This works under the assumption that rebalances
+that follow data shuffling terminate in less than 24 hours.
+
+**However**, in distributed systems, it is generally considered a bad practice
+to make assumptions that information propagates in a certain time interval:
+this consists in making a synchrony assumption, meaning that we are basically
+assuming a computing model that has much stronger properties than otherwise. To
+maximize the applicability of Garage, we would like to remove this assumption,
+and implement a system where time does not play a role. To do this, we would
+need to find a way to safely disable the GC when data is being shuffled around,
+and safely detect that the shuffling has terminated and thus the GC can be
+resumed. This introduces some complexity to the protocol and hasn't been
+tackled yet.
+
+### 2. Garbage collection of data blocks (in `data/` directory)
+
+Blocks in the data directory are reference-counted. In Garage versions before
+PR #135, blocks could get deleted from local disk as soon as their reference
+counter reached zero. We had a mechanism to not trigger this immediately at the
+rc-reaches-zero event, but the cleanup could be triggered by other means (for
+example by a block repair operation...). PR #135 added a safety measure so that
+blocks never get deleted in a 10 minute interval following the time when the RC
+reaches zero. This is a measure to make impossible race conditions such as #39.
+We would have liked to use a larger delay (e.g. 24 hours), but in the case of a
+rebalance of data, this would have led to the disk utilization to explode
+during the rebalancing, only to shrink again after 24 hours. The 10-minute
+delay is a compromise that gives good security while not having this problem of
+disk space explosion on rebalance.
 
-- CDC: <https://www.usenix.org/system/files/conference/atc16/atc16-paper-xia.pdf>
-- Erasure coding: <http://web.eecs.utk.edu/~jplank/plank/papers/CS-08-627.html>
-- [Openstack Storage Concepts](https://docs.openstack.org/arch-design/design-storage/design-storage-concepts.html)
-- [RADOS](https://ceph.com/wp-content/uploads/2016/08/weil-rados-pdsw07.pdf)

src/garage/admin.rs

@@ -446,7 +446,7 @@ impl AdminRpcHandler {
 		if opt.detailed {
 			writeln!(
 				&mut ret,
-				"  number of blocks: {}",
+				"  number of RC entries (~= number of blocks): {}",
 				self.garage.block_manager.rc_len()
 			)
 			.unwrap();

src/garage/cli/structs.rs

@@ -8,8 +8,7 @@ pub enum Command {
 	#[structopt(name = "server")]
 	Server,
 
-	/// Print identifier (public key) of this garage node.
+	/// Print identifier (public key) of this Garage node
-	/// Generates a new keypair if necessary.
 	#[structopt(name = "node-id")]
 	NodeId(NodeIdOpt),
 

src/model/block.rs

@@ -1,3 +1,4 @@
+use std::convert::TryInto;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
@@ -31,10 +32,20 @@ pub const INLINE_THRESHOLD: usize = 3072;
 pub const BACKGROUND_WORKERS: u64 = 1;
 pub const BACKGROUND_TRANQUILITY: u32 = 3;
 
-const BLOCK_RW_TIMEOUT: Duration = Duration::from_secs(42);
+// Timeout for RPCs that read and write blocks to remote nodes
-const BLOCK_GC_TIMEOUT: Duration = Duration::from_secs(60);
+const BLOCK_RW_TIMEOUT: Duration = Duration::from_secs(30);
+// Timeout for RPCs that ask other nodes whether they need a copy
+// of a given block before we delete it locally
 const NEED_BLOCK_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
-const RESYNC_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
+
+// The delay between the time where a resync operation fails
+// and the time when it is retried.
+const RESYNC_RETRY_DELAY: Duration = Duration::from_secs(60);
+
+// The delay between the moment when the reference counter
+// drops to zero, and the moment where we allow ourselves
+// to delete the block locally.
+const BLOCK_GC_DELAY: Duration = Duration::from_secs(600);
 
 /// RPC messages used to share blocks of data between nodes
 #[derive(Debug, Serialize, Deserialize)]
@@ -180,7 +191,7 @@ impl BlockManager {
 	/// that are required because of refcount > 0, and will try
 	/// to fix any mismatch between the two.
 	pub async fn repair_data_store(&self, must_exit: &watch::Receiver<bool>) -> Result<(), Error> {
-		// 1. Repair blocks from RC table
+		// 1. Repair blocks from RC table.
 		let garage = self.garage.load_full().unwrap();
 		let mut last_hash = None;
 		for (i, entry) in garage.block_ref_table.data.store.iter().enumerate() {
@@ -245,40 +256,51 @@ impl BlockManager {
 	/// Increment the number of time a block is used, putting it to resynchronization if it is
 	/// required, but not known
 	pub fn block_incref(&self, hash: &Hash) -> Result<(), Error> {
-		let old_rc = self.rc.fetch_and_update(&hash, |old| {
+		let old_rc = self
-			let old_v = old.map(u64_from_be_bytes).unwrap_or(0);
+			.rc
-			Some(u64::to_be_bytes(old_v + 1).to_vec())
+			.fetch_and_update(&hash, |old| RcEntry::parse_opt(old).increment().serialize())?;
-		})?;
+		let old_rc = RcEntry::parse_opt(old_rc);
-		let old_rc = old_rc.map(u64_from_be_bytes).unwrap_or(0);
+		if old_rc.is_zero() {
-		if old_rc == 0 {
+			// When the reference counter is incremented, there is
-			self.put_to_resync(hash, BLOCK_RW_TIMEOUT)?;
+			// normally a node that is responsible for sending us the
+			// data of the block. However that operation may fail,
+			// so in all cases we add the block here to the todo list
+			// to check later that it arrived correctly, and if not
+			// we will fecth it from someone.
+			self.put_to_resync(hash, 2 * BLOCK_RW_TIMEOUT)?;
 		}
 		Ok(())
 	}
 
 	/// Decrement the number of time a block is used
 	pub fn block_decref(&self, hash: &Hash) -> Result<(), Error> {
-		let new_rc = self.rc.update_and_fetch(&hash, |old| {
+		let new_rc = self
-			let old_v = old.map(u64_from_be_bytes).unwrap_or(0);
+			.rc
-			if old_v > 1 {
+			.update_and_fetch(&hash, |old| RcEntry::parse_opt(old).decrement().serialize())?;
-				Some(u64::to_be_bytes(old_v - 1).to_vec())
+		let new_rc = RcEntry::parse_opt(new_rc);
-			} else {
+		if let RcEntry::Deletable { .. } = new_rc {
-				None
+			self.put_to_resync(hash, BLOCK_GC_DELAY + Duration::from_secs(10))?;
-			}
-		})?;
-		if new_rc.is_none() {
-			self.put_to_resync(hash, BLOCK_GC_TIMEOUT)?;
 		}
 		Ok(())
 	}
 
 	/// Read a block's reference count
-	pub fn get_block_rc(&self, hash: &Hash) -> Result<u64, Error> {
+	fn get_block_rc(&self, hash: &Hash) -> Result<RcEntry, Error> {
-		Ok(self
+		Ok(RcEntry::parse_opt(self.rc.get(hash.as_ref())?))
-			.rc
+	}
-			.get(hash.as_ref())?
+
-			.map(u64_from_be_bytes)
+	/// Delete an entry in the RC table if it is deletable and the
-			.unwrap_or(0))
+	/// deletion time has passed
+	fn clear_deleted_block_rc(&self, hash: &Hash) -> Result<(), Error> {
+		let now = now_msec();
+		self.rc.update_and_fetch(&hash, |rcval| {
+			let updated = match RcEntry::parse_opt(rcval) {
+				RcEntry::Deletable { at_time } if now > at_time => RcEntry::Absent,
+				v => v,
+			};
+			updated.serialize()
+		})?;
+		Ok(())
 	}
 
 	// ---- Reading and writing blocks locally ----
@@ -300,7 +322,7 @@ impl BlockManager {
 			Ok(f) => f,
 			Err(e) => {
 				// Not found but maybe we should have had it ??
-				self.put_to_resync(hash, Duration::from_millis(0))?;
+				self.put_to_resync(hash, 2 * BLOCK_RW_TIMEOUT)?;
 				return Err(Into::into(e));
 			}
 		};
@@ -314,6 +336,7 @@ impl BlockManager {
 				.await
 				.move_block_to_corrupted(hash, self)
 				.await?;
+			self.put_to_resync(hash, Duration::from_millis(0))?;
 			return Err(Error::CorruptData(*hash));
 		}
 
@@ -328,7 +351,7 @@ impl BlockManager {
 			.await
 			.check_block_status(hash, self)
 			.await?;
-		Ok(needed && !exists)
+		Ok(needed.is_nonzero() && !exists)
 	}
 
 	/// Utility: gives the path of the directory in which a block should be found
@@ -349,7 +372,7 @@ impl BlockManager {
 	// ---- Resync loop ----
 
 	pub fn spawn_background_worker(self: Arc<Self>) {
-		// Launch 2 simultaneous workers for background resync loop preprocessing
+		// Launch n simultaneous workers for background resync loop preprocessing
 		for i in 0..BACKGROUND_WORKERS {
 			let bm2 = self.clone();
 			let background = self.system.background.clone();
@@ -400,14 +423,14 @@ impl BlockManager {
 
 	async fn resync_iter(&self, must_exit: &mut watch::Receiver<bool>) -> Result<bool, Error> {
 		if let Some((time_bytes, hash_bytes)) = self.resync_queue.pop_min()? {
-			let time_msec = u64_from_be_bytes(&time_bytes[0..8]);
+			let time_msec = u64::from_be_bytes(time_bytes[0..8].try_into().unwrap());
 			let now = now_msec();
 			if now >= time_msec {
 				let hash = Hash::try_from(&hash_bytes[..]).unwrap();
 				let res = self.resync_block(&hash).await;
 				if let Err(e) = &res {
 					warn!("Error when resyncing {:?}: {}", hash, e);
-					self.put_to_resync(&hash, RESYNC_RETRY_TIMEOUT)?;
+					self.put_to_resync(&hash, RESYNC_RETRY_DELAY)?;
 				}
 				Ok(true)
 			} else {
@@ -437,15 +460,18 @@ impl BlockManager {
 			.check_block_status(hash, self)
 			.await?;
 
-		if exists != needed {
+		if exists != needed.is_needed() || exists != needed.is_nonzero() {
-			info!(
+			debug!(
-				"Resync block {:?}: exists {}, needed {}",
+				"Resync block {:?}: exists {}, nonzero rc {}, deletable {}",
-				hash, exists, needed
+				hash,
+				exists,
+				needed.is_nonzero(),
+				needed.is_deletable(),
 			);
 		}
 
-		if exists && !needed {
+		if exists && needed.is_deletable() {
-			trace!("Offloading block {:?}", hash);
+			info!("Resync block {:?}: offloading and deleting", hash);
 
 			let mut who = self.replication.write_nodes(hash);
 			if who.len() < self.replication.write_quorum() {
@@ -488,7 +514,7 @@ impl BlockManager {
 					need_nodes.len()
 				);
 
-				let put_block_message = self.read_block(hash).await.err_context("PutBlock RPC")?;
+				let put_block_message = self.read_block(hash).await?;
 				self.system
 					.rpc
 					.try_call_many(
@@ -499,10 +525,11 @@ impl BlockManager {
 							.with_quorum(need_nodes.len())
 							.with_timeout(BLOCK_RW_TIMEOUT),
 					)
-					.await?;
+					.await
+					.err_context("PutBlock RPC")?;
 			}
 			info!(
-				"Deleting block {:?}, offload finished ({} / {})",
+				"Deleting unneeded block {:?}, offload finished ({} / {})",
 				hash,
 				need_nodes.len(),
 				who.len()
@@ -513,12 +540,16 @@ impl BlockManager {
 				.await
 				.delete_if_unneeded(hash, self)
 				.await?;
+
+			self.clear_deleted_block_rc(hash)?;
 		}
 
-		if needed && !exists {
+		if needed.is_nonzero() && !exists {
-			// TODO find a way to not do this if they are sending it to us
+			info!(
-			// Let's suppose this isn't an issue for now with the BLOCK_RW_TIMEOUT delay
+				"Resync block {:?}: fetching absent but needed block (refcount > 0)",
-			// between the RC being incremented and this part being called.
+				hash
+			);
+
 			let block_data = self.rpc_get_block(hash).await?;
 			self.write_block(hash, &block_data[..]).await?;
 		}
@@ -526,6 +557,8 @@ impl BlockManager {
 		Ok(())
 	}
 
+	// ---- Utility: iteration on files in the data directory ----
+
 	async fn for_each_file<F, Fut, State>(
 		&self,
 		state: State,
@@ -608,7 +641,7 @@ impl EndpointHandler<BlockRpc> for BlockManager {
 
 struct BlockStatus {
 	exists: bool,
-	needed: bool,
+	needed: RcEntry,
 }
 
 impl BlockManagerLocked {
@@ -620,7 +653,7 @@ impl BlockManagerLocked {
 		let path = mgr.block_path(hash);
 
 		let exists = fs::metadata(&path).await.is_ok();
-		let needed = mgr.get_block_rc(hash)? > 0;
+		let needed = mgr.get_block_rc(hash)?;
 
 		Ok(BlockStatus { exists, needed })
 	}
@@ -659,14 +692,13 @@ impl BlockManagerLocked {
 		let mut path2 = path.clone();
 		path2.set_extension("corrupted");
 		fs::rename(path, path2).await?;
-		mgr.put_to_resync(hash, Duration::from_millis(0))?;
 		Ok(())
 	}
 
 	async fn delete_if_unneeded(&self, hash: &Hash, mgr: &BlockManager) -> Result<(), Error> {
 		let BlockStatus { exists, needed } = self.check_block_status(hash, mgr).await?;
 
-		if exists && !needed {
+		if exists && needed.is_deletable() {
 			let path = mgr.block_path(hash);
 			fs::remove_file(path).await?;
 		}
@@ -674,9 +706,103 @@ impl BlockManagerLocked {
 	}
 }
 
-fn u64_from_be_bytes<T: AsRef<[u8]>>(bytes: T) -> u64 {
+/// Describes the state of the reference counter for a block
-	assert!(bytes.as_ref().len() == 8);
+#[derive(Clone, Copy, Debug)]
-	let mut x8 = [0u8; 8];
+enum RcEntry {
-	x8.copy_from_slice(bytes.as_ref());
+	/// Present: the block has `count` references, with `count` > 0.
-	u64::from_be_bytes(x8)
+	///
+	/// This is stored as u64::to_be_bytes(count)
+	Present { count: u64 },
+
+	/// Deletable: the block has zero references, and can be deleted
+	/// once time (returned by now_msec) is larger than at_time
+	/// (in millis since Unix epoch)
+	///
+	/// This is stored as [0u8; 8] followed by u64::to_be_bytes(at_time),
+	/// (this allows for the data format to be backwards compatible with
+	/// previous Garage versions that didn't have this intermediate state)
+	Deletable { at_time: u64 },
+
+	/// Absent: the block has zero references, and can be deleted
+	/// immediately
+	Absent,
+}
+
+impl RcEntry {
+	fn parse(bytes: &[u8]) -> Self {
+		if bytes.len() == 8 {
+			RcEntry::Present {
+				count: u64::from_be_bytes(bytes.try_into().unwrap()),
+			}
+		} else if bytes.len() == 16 {
+			RcEntry::Deletable {
+				at_time: u64::from_be_bytes(bytes[8..16].try_into().unwrap()),
+			}
+		} else {
+			panic!("Invalid RC entry: {:?}, database is corrupted. This is an error Garage is currently unable to recover from. Sorry, and also please report a bug.",
+				bytes
+			)
+		}
+	}
+
+	fn parse_opt<V: AsRef<[u8]>>(bytes: Option<V>) -> Self {
+		bytes
+			.map(|b| Self::parse(b.as_ref()))
+			.unwrap_or(Self::Absent)
+	}
+
+	fn serialize(self) -> Option<Vec<u8>> {
+		match self {
+			RcEntry::Present { count } => Some(u64::to_be_bytes(count).to_vec()),
+			RcEntry::Deletable { at_time } => {
+				Some([u64::to_be_bytes(0), u64::to_be_bytes(at_time)].concat())
+			}
+			RcEntry::Absent => None,
+		}
+	}
+
+	fn increment(self) -> Self {
+		let old_count = match self {
+			RcEntry::Present { count } => count,
+			_ => 0,
+		};
+		RcEntry::Present {
+			count: old_count + 1,
+		}
+	}
+
+	fn decrement(self) -> Self {
+		match self {
+			RcEntry::Present { count } => {
+				if count > 1 {
+					RcEntry::Present { count: count - 1 }
+				} else {
+					RcEntry::Deletable {
+						at_time: now_msec() + BLOCK_GC_DELAY.as_millis() as u64,
+					}
+				}
+			}
+			del => del,
+		}
+	}
+
+	fn is_zero(&self) -> bool {
+		matches!(self, RcEntry::Deletable { .. } | RcEntry::Absent)
+	}
+
+	fn is_nonzero(&self) -> bool {
+		!self.is_zero()
+	}
+
+	fn is_deletable(&self) -> bool {
+		match self {
+			RcEntry::Present { .. } => false,
+			RcEntry::Deletable { at_time } => now_msec() > *at_time,
+			RcEntry::Absent => true,
+		}
+	}
+
+	fn is_needed(&self) -> bool {
+		!self.is_deletable()
+	}
 }

src/table/data.rs

@@ -12,6 +12,7 @@ use garage_util::error::*;
 use garage_rpc::system::System;
 
 use crate::crdt::Crdt;
+use crate::gc::GcTodoEntry;
 use crate::replication::*;
 use crate::schema::*;
 
@@ -54,7 +55,7 @@ where
 			.expect("Unable to open DB Merkle TODO tree");
 
 		let gc_todo = db
-			.open_tree(&format!("{}:gc_todo", name))
+			.open_tree(&format!("{}:gc_todo_v2", name))
 			.expect("Unable to open DB tree");
 
 		Arc::new(Self {
@@ -176,7 +177,7 @@ where
 				let pk_hash = Hash::try_from(&tree_key[..32]).unwrap();
 				let nodes = self.replication.write_nodes(&pk_hash);
 				if nodes.first() == Some(&self.system.id) {
-					self.gc_todo.insert(&tree_key, new_bytes_hash.as_slice())?;
+					GcTodoEntry::new(tree_key, new_bytes_hash).save(&self.gc_todo)?;
 				}
 			}
 		}

src/table/gc.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::convert::TryInto;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -12,7 +13,8 @@ use futures_util::future::*;
 use tokio::sync::watch;
 
 use garage_util::data::*;
-use garage_util::error::Error;
+use garage_util::error::*;
+use garage_util::time::*;
 
 use garage_rpc::system::System;
 use garage_rpc::*;
@@ -24,7 +26,12 @@ use crate::schema::*;
 const TABLE_GC_BATCH_SIZE: usize = 1024;
 const TABLE_GC_RPC_TIMEOUT: Duration = Duration::from_secs(30);
 
-pub struct TableGc<F: TableSchema + 'static, R: TableReplication + 'static> {
+// GC delay for table entries: 1 day (24 hours)
+// (the delay before the entry is added in the GC todo list
+// and the moment the garbage collection actually happens)
+const TABLE_GC_DELAY: Duration = Duration::from_secs(24 * 3600);
+
+pub(crate) struct TableGc<F: TableSchema + 'static, R: TableReplication + 'static> {
 	system: Arc<System>,
 	data: Arc<TableData<F, R>>,
 
@@ -72,63 +79,100 @@ where
 	async fn gc_loop(self: Arc<Self>, mut must_exit: watch::Receiver<bool>) {
 		while !*must_exit.borrow() {
 			match self.gc_loop_iter().await {
-				Ok(true) => {
+				Ok(None) => {
 					// Stuff was done, loop immediately
-					continue;
 				}
-				Ok(false) => {
+				Ok(Some(wait_delay)) => {
-					// Nothing was done, sleep for some time (below)
+					// Nothing was done, wait specified delay.
+					select! {
+						_ = tokio::time::sleep(wait_delay).fuse() => {},
+						_ = must_exit.changed().fuse() => {},
+					}
 				}
 				Err(e) => {
 					warn!("({}) Error doing GC: {}", self.data.name, e);
 				}
 			}
-			select! {
-				_ = tokio::time::sleep(Duration::from_secs(10)).fuse() => {},
-				_ = must_exit.changed().fuse() => {},
-			}
 		}
 	}
 
-	async fn gc_loop_iter(&self) -> Result<bool, Error> {
+	async fn gc_loop_iter(&self) -> Result<Option<Duration>, Error> {
+		let now = now_msec();
+
 		let mut entries = vec![];
 		let mut excluded = vec![];
 
-		for item in self.data.gc_todo.iter() {
+		// List entries in the GC todo list
-			let (k, vhash) = item?;
+		// These entries are put there when a tombstone is inserted in the table
+		// (see update_entry in data.rs)
+		for entry_kv in self.data.gc_todo.iter() {
+			let (k, vhash) = entry_kv?;
+			let mut todo_entry = GcTodoEntry::parse(&k, &vhash);
+
+			if todo_entry.deletion_time() > now {
+				if entries.is_empty() && excluded.is_empty() {
+					// If the earliest entry in the todo list shouldn't yet be processed,
+					// return a duration to wait in the loop
+					return Ok(Some(Duration::from_millis(
+						todo_entry.deletion_time() - now,
+					)));
+				} else {
+					// Otherwise we have some entries to process, do a normal iteration.
+					break;
+				}
+			}
 
 			let vhash = Hash::try_from(&vhash[..]).unwrap();
 
-			let v_opt = self
+			// Check if the tombstone is still the current value of the entry.
+			// If not, we don't actually want to GC it, and we will remove it
+			// from the gc_todo table later (below).
+			todo_entry.value = self
 				.data
 				.store
 				.get(&k[..])?
-				.filter(|v| blake2sum(&v[..]) == vhash);
+				.filter(|v| blake2sum(&v[..]) == vhash)
+				.map(|v| v.to_vec());
 
-			if let Some(v) = v_opt {
+			if todo_entry.value.is_some() {
-				entries.push((ByteBuf::from(k.to_vec()), vhash, ByteBuf::from(v.to_vec())));
+				entries.push(todo_entry);
 				if entries.len() >= TABLE_GC_BATCH_SIZE {
 					break;
 				}
 			} else {
-				excluded.push((k, vhash));
+				excluded.push(todo_entry);
 			}
 		}
 
-		for (k, vhash) in excluded {
+		// Remove from gc_todo entries for tombstones where we have
-			self.todo_remove_if_equal(&k[..], vhash)?;
+		// detected that the current value has changed and
+		// is no longer a tombstone.
+		for entry in excluded {
+			entry.remove_if_equal(&self.data.gc_todo)?;
 		}
 
+		// Remaining in `entries` is the list of entries we want to GC,
+		// and for which they are still currently tombstones in the table.
+
 		if entries.is_empty() {
-			// Nothing to do in this iteration
+			// Nothing to do in this iteration (no entries present)
-			return Ok(false);
+			// Wait for a default delay of 60 seconds
+			return Ok(Some(Duration::from_secs(60)));
 		}
 
 		debug!("({}) GC: doing {} items", self.data.name, entries.len());
 
+		// Split entries to GC by the set of nodes on which they are stored.
+		// Here we call them partitions but they are not exactly
+		// the same as partitions as defined in the ring: those partitions
+		// are defined by the first 8 bits of the hash, but two of these
+		// partitions can be stored on the same set of nodes.
+		// Here we detect when entries are stored on the same set of nodes:
+		// even if they are not in the same 8-bit partition, we can still
+		// handle them together.
 		let mut partitions = HashMap::new();
-		for (k, vhash, v) in entries {
+		for entry in entries {
-			let pkh = Hash::try_from(&k[..32]).unwrap();
+			let pkh = Hash::try_from(&entry.key[..32]).unwrap();
 			let mut nodes = self.data.replication.write_nodes(&pkh);
 			nodes.retain(|x| *x != self.system.id);
 			nodes.sort();
@@ -136,9 +180,12 @@ where
 			if !partitions.contains_key(&nodes) {
 				partitions.insert(nodes.clone(), vec![]);
 			}
-			partitions.get_mut(&nodes).unwrap().push((k, vhash, v));
+			partitions.get_mut(&nodes).unwrap().push(entry);
 		}
 
+		// For each set of nodes that contains some items,
+		// ensure they are aware of the tombstone status, and once they
+		// are, instruct them to delete the entries.
 		let resps = join_all(
 			partitions
 				.into_iter()
@@ -146,6 +193,8 @@ where
 		)
 		.await;
 
+		// Collect errors and return a single error value even if several
+		// errors occurred.
 		let mut errs = vec![];
 		for resp in resps {
 			if let Err(e) = resp {
@@ -154,7 +203,7 @@ where
 		}
 
 		if errs.is_empty() {
-			Ok(true)
+			Ok(None)
 		} else {
 			Err(Error::Message(
 				errs.into_iter()
@@ -162,23 +211,42 @@ where
 					.collect::<Vec<_>>()
 					.join(", "),
 			))
+			.err_context("in try_send_and_delete in table GC:")
 		}
 	}
 
 	async fn try_send_and_delete(
 		&self,
 		nodes: Vec<Uuid>,
-		items: Vec<(ByteBuf, Hash, ByteBuf)>,
+		mut items: Vec<GcTodoEntry>,
 	) -> Result<(), Error> {
 		let n_items = items.len();
 
+		// Strategy: we first send all of the values to the remote nodes,
+		// to ensure that they are aware of the tombstone state,
+		// and that the previous state was correctly overwritten
+		// (if they have a newer state that overrides the tombstone, that's fine).
+		// Second, once everyone is at least at the tombstone state,
+		// we instruct everyone to delete the tombstone IF that is still their current state.
+		// If they are now at a different state, it means that that state overrides the
+		// tombstone in the CRDT lattice, and it will be propagated back to us at some point
+		// (either just a regular update that hasn't reached us yet, or later when the
+		// table is synced).
+
+		// Here, we store in updates all of the tombstones to send for step 1,
+		// and in deletes the list of keys and hashes of value for step 2.
 		let mut updates = vec![];
 		let mut deletes = vec![];
-		for (k, vhash, v) in items {
+		for item in items.iter_mut() {
-			updates.push(v);
+			updates.push(ByteBuf::from(item.value.take().unwrap()));
-			deletes.push((k, vhash));
+			deletes.push((ByteBuf::from(item.key.clone()), item.value_hash));
 		}
 
+		// Step 1: ensure everyone is at least at tombstone in CRDT lattice
+		// Here the quorum is nodes.len(): we cannot tolerate even a single failure,
+		// otherwise old values before the tombstone might come back in the data.
+		// GC'ing is not a critical function of the system, so it's not a big
+		// deal if we can't do it right now.
 		self.system
 			.rpc
 			.try_call_many(
@@ -189,40 +257,43 @@ where
 					.with_quorum(nodes.len())
 					.with_timeout(TABLE_GC_RPC_TIMEOUT),
 			)
-			.await?;
+			.await
+			.err_context("GC: send tombstones")?;
 
 		info!(
 			"({}) GC: {} items successfully pushed, will try to delete.",
 			self.data.name, n_items
 		);
 
+		// Step 2: delete tombstones everywhere.
+		// Here we also fail if even a single node returns a failure:
+		// it means that the garbage collection wasn't completed and has
+		// to be retried later.
 		self.system
 			.rpc
 			.try_call_many(
 				&self.endpoint,
 				&nodes[..],
-				GcRpc::DeleteIfEqualHash(deletes.clone()),
+				GcRpc::DeleteIfEqualHash(deletes),
 				RequestStrategy::with_priority(PRIO_BACKGROUND)
 					.with_quorum(nodes.len())
 					.with_timeout(TABLE_GC_RPC_TIMEOUT),
 			)
-			.await?;
+			.await
+			.err_context("GC: remote delete tombstones")?;
 
-		for (k, vhash) in deletes {
+		// GC has been successfull for all of these entries.
-			self.data.delete_if_equal_hash(&k[..], vhash)?;
+		// We now remove them all from our local table and from the GC todo list.
-			self.todo_remove_if_equal(&k[..], vhash)?;
+		for item in items {
+			self.data
+				.delete_if_equal_hash(&item.key[..], item.value_hash)
+				.err_context("GC: local delete tombstones")?;
+			item.remove_if_equal(&self.data.gc_todo)
+				.err_context("GC: remove from todo list after successfull GC")?;
 		}
 
 		Ok(())
 	}
-
-	fn todo_remove_if_equal(&self, key: &[u8], vhash: Hash) -> Result<(), Error> {
-		let _ = self
-			.data
-			.gc_todo
-			.compare_and_swap::<_, _, Vec<u8>>(key, Some(vhash), None)?;
-		Ok(())
-	}
 }
 
 #[async_trait]
@@ -240,7 +311,6 @@ where
 			GcRpc::DeleteIfEqualHash(items) => {
 				for (key, vhash) in items.iter() {
 					self.data.delete_if_equal_hash(&key[..], *vhash)?;
-					self.todo_remove_if_equal(&key[..], *vhash)?;
 				}
 				Ok(GcRpc::Ok)
 			}
@@ -248,3 +318,77 @@ where
 		}
 	}
 }
+
+/// An entry stored in the gc_todo Sled tree associated with the table
+/// Contains helper function for parsing, saving, and removing
+/// such entry in Sled
+///
+/// Format of an entry:
+/// - key =    8 bytes: timestamp of tombstone
+///                     (used to implement GC delay)
+///            n bytes: key in the main data table
+/// - value =  hash of the table entry to delete (the tombstone)
+///            for verification purpose, because we don't want to delete
+///            things that aren't tombstones
+pub(crate) struct GcTodoEntry {
+	tombstone_timestamp: u64,
+	key: Vec<u8>,
+	value_hash: Hash,
+	value: Option<Vec<u8>>,
+}
+
+impl GcTodoEntry {
+	/// Creates a new GcTodoEntry (not saved in Sled) from its components:
+	/// the key of an entry in the table, and the hash of the associated
+	/// serialized value
+	pub(crate) fn new(key: Vec<u8>, value_hash: Hash) -> Self {
+		Self {
+			tombstone_timestamp: now_msec(),
+			key,
+			value_hash,
+			value: None,
+		}
+	}
+
+	/// Parses a GcTodoEntry from a (k, v) pair stored in the gc_todo tree
+	pub(crate) fn parse(sled_k: &[u8], sled_v: &[u8]) -> Self {
+		Self {
+			tombstone_timestamp: u64::from_be_bytes(sled_k[0..8].try_into().unwrap()),
+			key: sled_k[8..].to_vec(),
+			value_hash: Hash::try_from(sled_v).unwrap(),
+			value: None,
+		}
+	}
+
+	/// Saves the GcTodoEntry in the gc_todo tree
+	pub(crate) fn save(&self, gc_todo_tree: &sled::Tree) -> Result<(), Error> {
+		gc_todo_tree.insert(self.todo_table_key(), self.value_hash.as_slice())?;
+		Ok(())
+	}
+
+	/// Removes the GcTodoEntry from the gc_todo tree if the
+	/// hash of the serialized value is the same here as in the tree.
+	/// This is usefull to remove a todo entry only under the condition
+	/// that it has not changed since the time it was read, i.e.
+	/// what we have to do is still the same
+	pub(crate) fn remove_if_equal(&self, gc_todo_tree: &sled::Tree) -> Result<(), Error> {
+		let _ = gc_todo_tree.compare_and_swap::<_, _, Vec<u8>>(
+			&self.todo_table_key()[..],
+			Some(self.value_hash),
+			None,
+		)?;
+		Ok(())
+	}
+
+	fn todo_table_key(&self) -> Vec<u8> {
+		[
+			&u64::to_be_bytes(self.tombstone_timestamp)[..],
+			&self.key[..],
+		]
+		.concat()
+	}
+
+	fn deletion_time(&self) -> u64 {
+		self.tombstone_timestamp + TABLE_GC_DELAY.as_millis() as u64
+	}
+}

src/util/config.rs

@@ -39,7 +39,7 @@ pub struct Config {
 	pub rpc_public_addr: Option<SocketAddr>,
 
 	/// Bootstrap peers RPC address
-	#[serde(deserialize_with = "deserialize_vec_addr")]
+	#[serde(deserialize_with = "deserialize_vec_addr", default)]
 	pub bootstrap_peers: Vec<(NodeID, SocketAddr)>,
 	/// Consul host to connect to to discover more peers
 	pub consul_host: Option<String>,