// SPDX-License-Identifier: Apache-2.0
//
// SchFS object layer, write side: create, link, unlink, rename, write,
// truncate.
//
// The same portable C as the read side, compiled both hosted (mkschfs) and
// freestanding (the kernel mount), so an image the tools produce and one the
// kernel produces come from a single implementation rather than two that agree
// by luck. Everything below is expressed as records in the one B-tree; there is
// no inode table, no directory block format and no bitmap to keep in step.
//
// Two rules run through the whole file.
//
// Nothing here may put a block on the stack. A SchFS block is 4096 bytes and so
// is the entire kernel stack, so every function borrows the volume's three
// scratch buffers (v->ctx) instead. Where a cursor is needed, ctx.a doubles as
// its buffer — which is safe only because no path below holds a cursor across a
// tree mutation, and every one of them copies what it needs out first.
//
// Nothing reachable from a committed root is ever overwritten. For tree blocks
// the B-tree enforces that; for *data* blocks it is enforced here, by
// schfs_extent.gen. A run allocated during the group being built may be written
// in place, because no published root can see it. Anything older is copied to a
// fresh block and the extent record split around it.

#include "schfs_int.h"

#ifdef SCHFS_HOSTED
#include <string.h>
#else
#include "types.h"
#include "defs.h"
#endif

/* ------------------------------------------------------------------ */
/* Inodes                                                              */
/* ------------------------------------------------------------------ */

uint64_t
schfs_alloc_objectid(struct schfs_vol *v)
{
    return v->next_objectid++;
}

/* ------------------------------------------------------------------ */
/* Snapshots                                                           */
/*                                                                     */
/* A snapshot is a tree root that is kept. Copy-on-write means the      */
/* blocks that root reaches are already immutable -- nothing overwrites */
/* a block a committed tree can reach -- so retaining one costs a       */
/* record, and keeping it alive is entirely schfs_space_scan's job.     */
/* ------------------------------------------------------------------ */

// Take a snapshot of the volume as it is committed *now*.
//
// The root recorded is the one in v->tree before this call, which is the
// committed state. It cannot be the root this call produces: adding the record
// rewrites the path to it under copy-on-write, so that root does not exist
// until afterwards, and a snapshot naming it would name a block nobody had
// written yet.
//
// The caller commits. Until it does, the record is not on disk and the
// snapshot does not exist -- which is the correct failure mode, because a
// snapshot that survived a crash while the tree it names did not would be a
// dangling root.
int
schfs_snap_create(struct schfs_vol *v, uint64_t id, const char *name,
                  unsigned namelen)
{
    schfs_cache_invalidate();

    if (namelen >= SCHFS_SNAP_NAME_MAX)
        return -1;

    struct schfs_snap sn;
    memset(&sn, 0, sizeof sn);
    sn.root            = v->tree.root;
    sn.root_crc        = v->tree.root_crc;
    sn.root_generation = v->tree.root_gen;
    sn.created         = v->tree.generation;
    for (unsigned i = 0; i < namelen; i++)
        sn.name[i] = name[i];

    struct schfs_key k;
    schfs_mkkey(&k, SCHFS_SNAP_OBJECTID, SCHFS_KEY_SNAP, id);
    int r = schfs_btree_insert(&v->tree, &v->ctx, &k, &sn, sizeof sn);
    if (r == 0)
        v->space.flags |= SCHFS_SPACE_RETAIN_ALL;
    return r;
}

// Forget a snapshot. The blocks it was holding are not freed here: they stop
// being reachable, and the next derivation at mount simply does not mark them.
// That is the whole deletion -- there is no reference count to decrement,
// because there never was one.
int
schfs_snap_delete(struct schfs_vol *v, uint64_t id)
{
    schfs_cache_invalidate();
    struct schfs_key k;
    schfs_mkkey(&k, SCHFS_SNAP_OBJECTID, SCHFS_KEY_SNAP, id);
    int r = schfs_btree_delete(&v->tree, &v->ctx, &k);
    if (r != 0)
        return r;

    // Stop retaining. If other snapshots remain, the next mount's derivation
    // sets the flag again -- it is the walk that knows, not this function, and
    // guessing here would mean keeping a second count to disagree with it.
    // Until that mount the allocator is merely conservative, which is the safe
    // direction.
    v->space.flags &= ~(unsigned)SCHFS_SPACE_RETAIN_ALL;
    return 0;
}

int
schfs_inode_put(struct schfs_vol *v, uint64_t ino, const struct schfs_inode *in)
{
    schfs_cache_invalidate();   /* any write can stale the read caches in obj.c */
    struct schfs_key k;
    schfs_mkkey(&k, ino, SCHFS_KEY_INODE, 0);
    return schfs_btree_insert(&v->tree, &v->ctx, &k, in, sizeof(*in));
}

// Read an inode using the volume's own scratch block. Every mutation below is a
// read-modify-write of one of these, and each one that took a cursor buffer
// from its caller was one more place to get the aliasing rules wrong.
static int
inode_load(struct schfs_vol *v, uint64_t ino, struct schfs_inode *out)
{
    struct schfs_cursor c;
    schfs_cursor_init(&c, &v->tree, v->ctx.a);
    return schfs_inode_get(v, &c, ino, out);
}

/* ------------------------------------------------------------------ */
/* Directory entries                                                   */
/*                                                                     */
/* One key per name hash, holding a packed list of records, because two */
/* names in a directory can hash the same and the second insert would   */
/* otherwise silently replace the first.                                */
/* ------------------------------------------------------------------ */

// Load the packed list for `name`'s hash into v->dirbuf. Returns its length, or
// 0 if there is no such key, or a negative error.
static int
dirlist_load(struct schfs_vol *v, uint64_t dir, const char *name,
             unsigned namelen, struct schfs_key *k)
{
    schfs_mkkey(k, dir, SCHFS_KEY_DIRENT, schfs_namehash(name, namelen));

    unsigned len = sizeof(v->dirbuf);
    int r = schfs_btree_lookup(&v->tree, &v->ctx, k, v->dirbuf, &len);
    if (r == -2)
        return -2;
    if (r != 0)
        return 0;
    if (len > sizeof(v->dirbuf))
        return -2;                   /* stored value larger than can exist */
    return (int)len;
}

// Offset of `name` within the packed list, or -1 if absent, -2 if malformed.
static int
dirlist_find(const char *list, unsigned len, const char *name, unsigned namelen)
{
    for (unsigned off = 0; off + sizeof(struct schfs_dirent_rec) <= len; ) {
        const struct schfs_dirent_rec *d =
            (const struct schfs_dirent_rec *)(list + off);
        unsigned step = SCHFS_DIRENT_LEN(d->namelen);
        if (step == 0 || off + step > len)
            return -2;
        if (d->namelen == namelen &&
            memcmp(list + off + sizeof(*d), name, namelen) == 0)
            return (int)off;
        off += step;
    }
    return -1;
}

int
schfs_dir_add(struct schfs_vol *v, uint64_t dir, const char *name,
              unsigned namelen, uint64_t child, uint16_t type)
{
    schfs_cache_invalidate();   /* any write can stale the read caches in obj.c */
    if (namelen == 0 || namelen > SCHFS_NAME_MAX)
        return -1;

    struct schfs_key k;
    int got = dirlist_load(v, dir, name, namelen, &k);
    if (got < 0)
        return got;
    unsigned len = (unsigned)got;

    // Reject a duplicate outright. Without this, adding a name twice produces a
    // directory with two entries that lookup can never tell apart.
    int at = dirlist_find(v->dirbuf, len, name, namelen);
    if (at == -2)
        return -2;
    if (at >= 0)
        return -1;

    unsigned add = SCHFS_DIRENT_LEN(namelen);
    if (len + add > sizeof(v->dirbuf))
        return -4;                   /* collision list will not fit in one item */

    struct schfs_dirent_rec *d = (struct schfs_dirent_rec *)(v->dirbuf + len);
    memset(d, 0, add);
    d->objectid = child;
    d->type     = type;
    d->namelen  = (uint16_t)namelen;
    memmove(v->dirbuf + len + sizeof(*d), name, namelen);

    return schfs_btree_insert(&v->tree, &v->ctx, &k, v->dirbuf, len + add);
}

int
schfs_dir_del(struct schfs_vol *v, uint64_t dir, const char *name,
              unsigned namelen)
{
    schfs_cache_invalidate();   /* any write can stale the read caches in obj.c */
    if (namelen == 0 || namelen > SCHFS_NAME_MAX)
        return -1;

    struct schfs_key k;
    int got = dirlist_load(v, dir, name, namelen, &k);
    if (got < 0)
        return got;
    unsigned len = (unsigned)got;
    if (len == 0)
        return -1;

    int at = dirlist_find(v->dirbuf, len, name, namelen);
    if (at == -2)
        return -2;
    if (at < 0)
        return -1;

    const struct schfs_dirent_rec *d =
        (const struct schfs_dirent_rec *)(v->dirbuf + at);
    unsigned step = SCHFS_DIRENT_LEN(d->namelen);
    memmove(v->dirbuf + at, v->dirbuf + at + step, len - (unsigned)at - step);
    len -= step;

    // An emptied list means the key itself goes, not a zero-length value: a key
    // that exists with nothing in it would make the directory scan stop on a
    // record readdir cannot describe.
    if (len == 0)
        return schfs_btree_delete(&v->tree, &v->ctx, &k);
    return schfs_btree_insert(&v->tree, &v->ctx, &k, v->dirbuf, len);
}

int
schfs_dir_empty(struct schfs_vol *v, uint64_t dir)
{
    struct schfs_key start;
    schfs_mkkey(&start, dir, SCHFS_KEY_DIRENT, 0);

    struct schfs_cursor c;
    schfs_cursor_init(&c, &v->tree, v->ctx.a);
    int r = schfs_cursor_seek(&c, &start);
    if (r == -2)
        return -2;
    if (r != 0)
        return 1;                    /* nothing at or after: empty */

    const struct schfs_key *got = schfs_cursor_key(&c);
    if (!got)
        return 1;
    return (got->objectid == dir && got->type == SCHFS_KEY_DIRENT) ? 0 : 1;
}

/* ------------------------------------------------------------------ */
/* Data-block checksums                                                */
/*                                                                     */
/* Tree nodes carry their checksum in their parent's pointer. Data      */
/* blocks have no parent, so theirs live under one reserved object id,  */
/* exactly as btrfs keeps a checksum tree.                              */
/* ------------------------------------------------------------------ */

static int
csum_put(struct schfs_vol *v, uint64_t dblk, const void *data)
{
    struct schfs_key k;
    schfs_mkkey(&k, SCHFS_CSUM_OBJECTID, SCHFS_KEY_CSUM, dblk);
    uint32_t c = schfs_crc32c(0, data, SCHFS_BLKSIZE);
    return schfs_btree_insert(&v->tree, &v->ctx, &k, &c, sizeof(c));
}

static int
csum_del(struct schfs_vol *v, uint64_t dblk)
{
    struct schfs_key k;
    schfs_mkkey(&k, SCHFS_CSUM_OBJECTID, SCHFS_KEY_CSUM, dblk);
    int r = schfs_btree_delete(&v->tree, &v->ctx, &k);
    return r == -1 ? 0 : r;          /* absent is not a failure */
}

/* ------------------------------------------------------------------ */
/* Extents                                                             */
/* ------------------------------------------------------------------ */

static int
extent_put(struct schfs_vol *v, uint64_t ino, uint64_t fblk,
           const struct schfs_extent *e)
{
    struct schfs_key k;
    schfs_mkkey(&k, ino, SCHFS_KEY_EXTENT, fblk);
    return schfs_btree_insert(&v->tree, &v->ctx, &k, e, sizeof(*e));
}

static int
extent_del(struct schfs_vol *v, uint64_t ino, uint64_t fblk)
{
    struct schfs_key k;
    schfs_mkkey(&k, ino, SCHFS_KEY_EXTENT, fblk);
    int r = schfs_btree_delete(&v->tree, &v->ctx, &k);
    return r == -1 ? 0 : r;
}

// The extent covering file block `fblk`, if any. Extents are keyed by the file
// block they start at, so the record governing block B is the last one at or
// before (ino, EXTENT, B). Returns 1 and fills *start and *e, 0 for a hole, -2
// on corruption.
static int
extent_find(struct schfs_vol *v, uint64_t ino, uint64_t fblk,
            uint64_t *start, struct schfs_extent *e)
{
    struct schfs_cursor c;
    schfs_cursor_init(&c, &v->tree, v->ctx.a);

    struct schfs_key probe;
    schfs_mkkey(&probe, ino, SCHFS_KEY_EXTENT, fblk);
    int r = schfs_btree_lookup_le(&c, &probe);
    if (r == -2)
        return -2;
    if (r != 0)
        return 0;

    const struct schfs_key *got = schfs_cursor_key(&c);
    unsigned vlen;
    const struct schfs_extent *found =
        (const struct schfs_extent *)schfs_cursor_val(&c, &vlen);
    if (!got || !found)
        return 0;
    // lookup_le happily lands on some other object's records; so does a sparse
    // file whose first extent starts past this block.
    if (got->objectid != ino || got->type != SCHFS_KEY_EXTENT)
        return 0;
    if (vlen != sizeof(*e))
        return -2;
    if (fblk >= got->offset + found->nblocks)
        return 0;

    *start = got->offset;
    *e     = *found;
    return 1;
}

// The last extent record belonging to `ino`, or 0 if it has none.
static int
extent_last(struct schfs_vol *v, uint64_t ino, uint64_t *start,
            struct schfs_extent *e)
{
    struct schfs_cursor c;
    schfs_cursor_init(&c, &v->tree, v->ctx.a);

    struct schfs_key probe;
    schfs_mkkey(&probe, ino, SCHFS_KEY_EXTENT, ~0ULL);
    int r = schfs_btree_lookup_le(&c, &probe);
    if (r == -2)
        return -2;
    if (r != 0)
        return 0;

    const struct schfs_key *got = schfs_cursor_key(&c);
    unsigned vlen;
    const struct schfs_extent *found =
        (const struct schfs_extent *)schfs_cursor_val(&c, &vlen);
    if (!got || !found)
        return 0;
    if (got->objectid != ino || got->type != SCHFS_KEY_EXTENT)
        return 0;
    if (vlen != sizeof(*e))
        return -2;

    *start = got->offset;
    *e     = *found;
    return 1;
}

// Record a newly allocated single block, merging it into the preceding extent
// when the two are adjacent both in the file and on the disk.
//
// This is what keeps extents worth having. Sequential writes get consecutive
// blocks from the allocator, so without merging a one-megabyte file carries 256
// extent records instead of one, and the tree ends up mostly describing where
// the data is rather than holding it.
static int
extent_grow(struct schfs_vol *v, uint64_t ino, uint64_t fblk,
            const struct schfs_extent *e)
{
    if (fblk > 0) {
        uint64_t pstart;
        struct schfs_extent p;
        int m = extent_find(v, ino, fblk - 1, &pstart, &p);
        if (m == -2)
            return -2;
        if (m == 1 &&
            pstart + p.nblocks == fblk &&
            p.disk_block + p.nblocks == e->disk_block &&
            p.refs == e->refs && p.gen == e->gen) {
            p.nblocks++;
            return extent_put(v, ino, pstart, &p);
        }
    }
    return extent_put(v, ino, fblk, e);
}

// Release a run of data blocks and the checksum records that describe them.
// Shared runs are left alone: another file still points at them, and proper
// refcounting needs an extent tree that nothing creates yet.
static int
extent_release(struct schfs_vol *v, const struct schfs_extent *e,
               uint64_t first, uint64_t n)
{
    if (e->refs != 1)
        return 0;
    for (uint64_t i = 0; i < n; i++) {
        int r = csum_del(v, first + i);
        if (r != 0)
            return r;
        v->dev.bfree(v->dev.ctx, first + i);
    }
    return 0;
}

// Drop everything from file block `from` onward. Used by truncate and by the
// deletion of an object.
static int
extents_drop_from(struct schfs_vol *v, uint64_t ino, struct schfs_inode *in,
                  uint64_t from)
{
    for (;;) {
        uint64_t start;
        struct schfs_extent e;
        int m = extent_last(v, ino, &start, &e);
        if (m == -2)
            return -2;
        if (m == 0)
            return 0;                /* no extents left */
        if (start + e.nblocks <= from)
            return 0;                /* nothing beyond the cut */

        if (start >= from) {
            int r = extent_release(v, &e, e.disk_block, e.nblocks);
            if (r != 0)
                return r;
            r = extent_del(v, ino, start);
            if (r != 0)
                return r;
            if (in->blocks >= e.nblocks)
                in->blocks -= e.nblocks;
            continue;                /* there may be earlier ones too */
        }

        // The cut falls inside this run: keep its head, release its tail.
        uint64_t keep = from - start;
        uint64_t tail = e.nblocks - keep;
        int r = extent_release(v, &e, e.disk_block + keep, tail);
        if (r != 0)
            return r;
        e.nblocks = keep;
        r = extent_put(v, ino, start, &e);
        if (r != 0)
            return r;
        if (in->blocks >= tail)
            in->blocks -= tail;
        return 0;
    }
}

/* ------------------------------------------------------------------ */
/* File data                                                           */
/* ------------------------------------------------------------------ */

int
schfs_file_reserve(struct schfs_vol *v, uint64_t ino, uint64_t nblocks)
{
    if (nblocks == 0)
        return 0;

    // Take the whole run before touching the tree.
    //
    // This is what makes extents worth having when the final size is already
    // known. Allocating a data block, then inserting its extent record -- which
    // itself allocates blocks for the copy-on-write path -- then allocating the
    // next data block interleaves metadata with data, so consecutive file
    // blocks are never physically adjacent. Grabbing the run up front keeps it
    // contiguous and a whole file becomes one record.
    uint64_t first;
    if (v->dev.balloc_run) {
        first = v->dev.balloc_run(v->dev.ctx, nblocks);
        if (first == 0)
            return -3;
    } else {
        first = v->dev.balloc(v->dev.ctx);
        if (first == 0)
            return -3;
        uint64_t prev = first;
        for (uint64_t i = 1; i < nblocks; i++) {
            uint64_t b = v->dev.balloc(v->dev.ctx);
            if (b == 0)
                return -3;
            if (b != prev + 1)
                return -2;           /* allocator is not handing out a run */
            prev = b;
        }
    }

    struct schfs_extent e;
    memset(&e, 0, sizeof(e));
    e.disk_block = first;
    e.nblocks    = nblocks;
    e.refs       = 1;
    e.gen        = v->tree.generation;

    // Deliberately does not touch the inode. Re-reading and rewriting it here
    // costs two extra tree inserts per file, and every insert copies its whole
    // path. The caller knows the block count before it writes the inode the
    // first time, so it sets in.blocks there instead.
    return extent_put(v, ino, 0, &e);
}

// Where the next write to file block `fblk` should land.
struct wblk {
    uint64_t disk;      // block to write
    uint64_t src;       // block holding the current contents, 0 for a hole
    int      fresh;     // a block that did not exist before
};

static int
block_for_write(struct schfs_vol *v, uint64_t ino, uint64_t fblk,
                struct wblk *w)
{
    uint64_t start;
    struct schfs_extent e;
    int m = extent_find(v, ino, fblk, &start, &e);
    if (m == -2)
        return -2;

    if (m == 0) {
        uint64_t d = v->dev.balloc(v->dev.ctx);
        if (d == 0)
            return -3;

        struct schfs_extent ne;
        memset(&ne, 0, sizeof(ne));
        ne.disk_block = d;
        ne.nblocks    = 1;
        ne.refs       = 1;
        ne.gen        = v->tree.generation;

        int r = extent_grow(v, ino, fblk, &ne);
        if (r != 0)
            return r;

        w->disk = d;
        w->src = 0;
        w->fresh = 1;
        return 0;
    }

    uint64_t old = e.disk_block + (fblk - start);

    // Allocated by the group being built, and owned outright: no published root
    // can reach it, so it may simply be overwritten.
    if (e.gen == v->tree.generation && e.refs == 1) {
        w->disk = old;
        w->src = old;
        w->fresh = 0;
        return 0;
    }

    // Otherwise copy this one block out of the run, and split the record around
    // it: a head covering what came before, the new block, and a tail covering
    // what came after. No deletion is needed — when the block being replaced is
    // the run's first, the middle record simply overwrites the original key.
    uint64_t nd = v->dev.balloc(v->dev.ctx);
    if (nd == 0)
        return -3;

    if (fblk > start) {
        struct schfs_extent head = e;
        head.nblocks = fblk - start;
        int r = extent_put(v, ino, start, &head);
        if (r != 0)
            return r;
    }
    {
        struct schfs_extent mid;
        memset(&mid, 0, sizeof(mid));
        mid.disk_block = nd;
        mid.nblocks    = 1;
        mid.refs       = 1;
        mid.gen        = v->tree.generation;
        int r = extent_put(v, ino, fblk, &mid);
        if (r != 0)
            return r;
    }
    if (fblk + 1 < start + e.nblocks) {
        struct schfs_extent tail = e;
        tail.disk_block = e.disk_block + (fblk + 1 - start);
        tail.nblocks    = start + e.nblocks - (fblk + 1);
        int r = extent_put(v, ino, fblk + 1, &tail);
        if (r != 0)
            return r;
    }

    // The block just replaced rejoins the free map at the next commit, not now:
    // until then the committed tree still points at it and a crash lands there.
    int r = extent_release(v, &e, old, 1);
    if (r != 0)
        return r;

    w->disk = nd;
    w->src = old;
    w->fresh = 0;
    return 0;
}

long
schfs_file_write(struct schfs_vol *v, uint64_t ino, uint64_t off,
                 const void *buf, unsigned n, void *dblk)
{
    schfs_cache_invalidate();   /* any write can stale the read caches in obj.c */
    struct schfs_inode in;
    int r = inode_load(v, ino, &in);
    if (r != 0)
        return r;

    const char *src  = (const char *)buf;
    unsigned    done = 0;

    while (done < n) {
        uint64_t at    = off + done;
        uint64_t fblk  = at / SCHFS_BLKSIZE;
        unsigned inblk = (unsigned)(at % SCHFS_BLKSIZE);
        unsigned want  = SCHFS_BLKSIZE - inblk;
        if (want > n - done)
            want = n - done;

        struct wblk w;
        r = block_for_write(v, ino, fblk, &w);
        if (r != 0)
            return r;
        if (w.fresh)
            in.blocks++;

        // A whole-block write needs no read: every byte is about to be
        // replaced. A partial one has to start from whatever is there, or from
        // zeros where the file had a hole.
        if (inblk != 0 || want != SCHFS_BLKSIZE) {
            if (w.src) {
                if (v->dev.bread(v->dev.ctx, w.src, dblk) != 0)
                    return -2;
            } else {
                memset(dblk, 0, SCHFS_BLKSIZE);
            }
        }
        memmove((char *)dblk + inblk, src + done, want);

        if (v->dev.bwrite(v->dev.ctx, w.disk, dblk) != 0)
            return -2;
        // The checksum is rewritten every time the block changes, not just when
        // it is allocated, or a topped-up tail block fails its own check on the
        // very next read.
        r = csum_put(v, w.disk, dblk);
        if (r != 0)
            return r;

        done += want;
    }

    if (off + n > in.size)
        in.size = off + n;
    in.gen = v->tree.generation;
    r = schfs_inode_put(v, ino, &in);
    if (r != 0)
        return r;
    return (long)done;
}

int
schfs_file_append(struct schfs_vol *v, uint64_t ino, const void *buf,
                  unsigned n, void *dblk)
{
    struct schfs_inode in;
    int r = inode_load(v, ino, &in);
    if (r != 0)
        return r == -1 ? -2 : r;

    long got = schfs_file_write(v, ino, in.size, buf, n, dblk);
    if (got < 0)
        return (int)got;
    return got == (long)n ? 0 : -2;
}

int
schfs_file_truncate(struct schfs_vol *v, uint64_t ino, uint64_t size,
                    void *dblk)
{
    schfs_cache_invalidate();   /* any write can stale the read caches in obj.c */
    struct schfs_inode in;
    int r = inode_load(v, ino, &in);
    if (r != 0)
        return r;

    if (size < in.size) {
        uint64_t keep = (size + SCHFS_BLKSIZE - 1) / SCHFS_BLKSIZE;
        r = extents_drop_from(v, ino, &in, keep);
        if (r != 0)
            return r;

        // Whole blocks past the new end are gone; the block the new size falls
        // *inside* has to have its tail zeroed. It is tempting not to bother,
        // since a read stops at in.size and can never see those bytes — but
        // truncating down and then growing again is required to read back
        // zeros, and without this it reads back whatever the file used to hold.
        unsigned tail = (unsigned)(size % SCHFS_BLKSIZE);
        if (tail != 0) {
            uint64_t fblk = size / SCHFS_BLKSIZE;
            uint64_t start;
            struct schfs_extent e;
            int m = extent_find(v, ino, fblk, &start, &e);
            if (m == -2)
                return -2;
            if (m == 1) {            /* a hole is already zeros */
                struct wblk w;
                r = block_for_write(v, ino, fblk, &w);
                if (r != 0)
                    return r;
                if (w.src) {
                    if (v->dev.bread(v->dev.ctx, w.src, dblk) != 0)
                        return -2;
                } else {
                    memset(dblk, 0, SCHFS_BLKSIZE);
                }
                memset((char *)dblk + tail, 0, SCHFS_BLKSIZE - tail);
                if (v->dev.bwrite(v->dev.ctx, w.disk, dblk) != 0)
                    return -2;
                r = csum_put(v, w.disk, dblk);
                if (r != 0)
                    return r;
                if (w.fresh)
                    in.blocks++;
            }
        }
    }
    // Growing needs nothing at all: the range with no extent records reads back
    // as zeros, which is a sparse file for free.

    in.size = size;
    in.gen  = v->tree.generation;
    return schfs_inode_put(v, ino, &in);
}

// Release an object entirely: its data, its checksums and its inode record.
int
schfs_object_free(struct schfs_vol *v, uint64_t ino)
{
    struct schfs_inode in;
    int r = inode_load(v, ino, &in);
    if (r == -1)
        return 0;                    /* already gone */
    if (r != 0)
        return r;

    r = extents_drop_from(v, ino, &in, 0);
    if (r != 0)
        return r;

    struct schfs_key k;
    schfs_mkkey(&k, ino, SCHFS_KEY_INODE, 0);
    r = schfs_btree_delete(&v->tree, &v->ctx, &k);
    return r == -1 ? 0 : r;
}

/* ------------------------------------------------------------------ */
/* Namespace                                                           */
/* ------------------------------------------------------------------ */

// A directory's link count is 2 (itself and its own ".") plus one per
// subdirectory, because each subdirectory's ".." points back at it. Neither
// entry is stored, but the count still has to match what POSIX tools expect or
// "find -type d" prunes real directories and du walks the tree wrongly.
static int
parent_nlink_adjust(struct schfs_vol *v, uint64_t dir, int delta)
{
    struct schfs_inode in;
    int r = inode_load(v, dir, &in);
    if (r != 0)
        return r;
    if (delta < 0 && in.nlink < (uint32_t)(-delta))
        return -2;
    in.nlink = (uint32_t)((int64_t)in.nlink + delta);
    return schfs_inode_put(v, dir, &in);
}

int
schfs_create(struct schfs_vol *v, uint64_t dir, const char *name,
             unsigned namelen, uint16_t type, uint32_t mode,
             uint16_t rdev_major, uint16_t rdev_minor, uint64_t now,
             uint64_t *out_ino)
{
    if (namelen == 0 || namelen > SCHFS_NAME_MAX)
        return -1;

    struct schfs_inode parent;
    int r = inode_load(v, dir, &parent);
    if (r != 0)
        return r;
    if (parent.type != SCHFS_T_DIR)
        return -1;

    uint64_t ino = schfs_alloc_objectid(v);

    struct schfs_inode in;
    memset(&in, 0, sizeof(in));
    in.type       = type;
    in.parent     = dir;
    in.mode       = mode;
    in.nlink      = type == SCHFS_T_DIR ? 2 : 1;
    in.rdev_major = rdev_major;
    in.rdev_minor = rdev_minor;
    in.atime = in.mtime = in.ctime = now;
    in.gen        = v->tree.generation;

    r = schfs_inode_put(v, ino, &in);
    if (r != 0)
        return r;

    // The entry goes in last, so a failure leaves an unreferenced object rather
    // than a name pointing at nothing. Neither survives a crash — none of this
    // is visible until the transaction commits — but an unreachable inode is
    // the harmless one of the two if a later step in the same group fails.
    r = schfs_dir_add(v, dir, name, namelen, ino, type);
    if (r != 0)
        return r;

    if (type == SCHFS_T_DIR) {
        r = parent_nlink_adjust(v, dir, +1);
        if (r != 0)
            return r;
    }

    if (out_ino)
        *out_ino = ino;
    return 0;
}

int
schfs_link(struct schfs_vol *v, uint64_t dir, const char *name,
           unsigned namelen, uint64_t ino)
{
    struct schfs_inode in;
    int r = inode_load(v, ino, &in);
    if (r != 0)
        return r;
    // Directories are not hard-linkable. Allowing it makes cycles possible, and
    // a cycle turns every recursive walk into an infinite one.
    if (in.type == SCHFS_T_DIR)
        return -1;

    r = schfs_dir_add(v, dir, name, namelen, ino, in.type);
    if (r != 0)
        return r;

    in.nlink++;
    in.ctime = in.mtime;
    return schfs_inode_put(v, ino, &in);
}

int
schfs_unlink(struct schfs_vol *v, uint64_t dir, const char *name,
             unsigned namelen)
{
    uint64_t ino;
    uint16_t type;

    {
        struct schfs_cursor c;
        schfs_cursor_init(&c, &v->tree, v->ctx.a);
        int r = schfs_dir_lookup(v, &c, dir, name, namelen, &ino, &type);
        if (r != 0)
            return r;
    }

    struct schfs_inode in;
    int r = inode_load(v, ino, &in);
    if (r != 0)
        return r;

    if (in.type == SCHFS_T_DIR) {
        int e = schfs_dir_empty(v, ino);
        if (e == -2)
            return -2;
        if (e == 0)
            return -5;               /* directory not empty */
    }

    r = schfs_dir_del(v, dir, name, namelen);
    if (r != 0)
        return r;

    if (in.type == SCHFS_T_DIR) {
        r = parent_nlink_adjust(v, dir, -1);
        if (r != 0)
            return r;
        return schfs_object_free(v, ino);
    }

    if (in.nlink > 1) {
        in.nlink--;
        return schfs_inode_put(v, ino, &in);
    }
    return schfs_object_free(v, ino);
}

int
schfs_rename(struct schfs_vol *v, uint64_t olddir, const char *oldname,
             unsigned oldlen, uint64_t newdir, const char *newname,
             unsigned newlen)
{
    uint64_t ino;
    uint16_t type;

    {
        struct schfs_cursor c;
        schfs_cursor_init(&c, &v->tree, v->ctx.a);
        int r = schfs_dir_lookup(v, &c, olddir, oldname, oldlen, &ino, &type);
        if (r != 0)
            return r;
    }

    // Renaming a directory into its own subtree would detach that subtree from
    // the root and leave it reachable only from itself. Walking up from the new
    // parent is the cheap way to catch it, since every directory records its
    // own parent.
    if (type == SCHFS_T_DIR && newdir != olddir) {
        uint64_t up = newdir;
        for (int i = 0; i <= SCHFS_MAX_LEVEL * 8 && up != 0; i++) {
            if (up == ino)
                return -1;
            struct schfs_inode t;
            if (inode_load(v, up, &t) != 0)
                break;
            if (up == v->root_objectid)
                break;
            up = t.parent;
        }
    }

    // A target that already exists is replaced, which means removing it first —
    // and refusing when that would silently discard a non-empty directory.
    {
        struct schfs_cursor c;
        schfs_cursor_init(&c, &v->tree, v->ctx.a);
        uint64_t victim;
        if (schfs_dir_lookup(v, &c, newdir, newname, newlen, &victim, 0) == 0) {
            if (victim == ino)
                return 0;            /* renaming a name onto itself */
            int r = schfs_unlink(v, newdir, newname, newlen);
            if (r != 0)
                return r;
        }
    }

    int r = schfs_dir_add(v, newdir, newname, newlen, ino, type);
    if (r != 0)
        return r;
    r = schfs_dir_del(v, olddir, oldname, oldlen);
    if (r != 0)
        return r;

    if (newdir != olddir) {
        if (type == SCHFS_T_DIR) {
            // ".." is synthesised from the inode's parent field, so moving a
            // directory means rewriting that field and moving the link its ".."
            // contributes to each parent's count.
            struct schfs_inode in;
            r = inode_load(v, ino, &in);
            if (r != 0)
                return r;
            in.parent = newdir;
            r = schfs_inode_put(v, ino, &in);
            if (r != 0)
                return r;
            r = parent_nlink_adjust(v, olddir, -1);
            if (r != 0)
                return r;
            r = parent_nlink_adjust(v, newdir, +1);
            if (r != 0)
                return r;
        }
    }
    return 0;
}