Add SCOUTFS_IOC_RECOMPUTE_TOTL ioctl.

Walk every contributing xattr for a given totl key, sum the values
and keep count. Then, read the merged value via wkic, and apply a
corrective delta.  If a delta is found, calls scoutfs_warning()
so that system logs capture the correction values.

Signed-off-by: Auke Kok <auke.kok@versity.com>

kmod/src/ioctl.c

@@ -1776,6 +1776,21 @@ out:
 	return ret;
 }
 
+static long scoutfs_ioc_recompute_totl(struct file *file, unsigned long arg)
+{
+	struct super_block *sb = file_inode(file)->i_sb;
+	struct scoutfs_ioctl_recompute_totl __user *urt = (void __user *)arg;
+	struct scoutfs_ioctl_recompute_totl rt;
+
+	if (!capable(CAP_SYS_ADMIN))
+		return -EPERM;
+
+	if (copy_from_user(&rt, urt, sizeof(rt)))
+		return -EFAULT;
+
+	return scoutfs_xattr_recompute_totl(sb, rt.name);
+}
+
 long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 {
 	switch (cmd) {
@@ -1829,6 +1844,8 @@ long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		return scoutfs_ioc_punch_offline(file, arg);
 	case SCOUTFS_IOC_INJECT_TOTL_DELTA:
 		return scoutfs_ioc_inject_totl_delta(file, arg);
+	case SCOUTFS_IOC_RECOMPUTE_TOTL:
+		return scoutfs_ioc_recompute_totl(file, arg);
 	}
 
 	return -ENOTTY;

kmod/src/ioctl.h

@@ -889,4 +889,16 @@ struct scoutfs_ioctl_inject_totl_delta {
 #define SCOUTFS_IOC_INJECT_TOTL_DELTA \
 	_IOW(SCOUTFS_IOCTL_MAGIC, 25, struct scoutfs_ioctl_inject_totl_delta)
 
+/*
+ * Recompute the totl item @name identifies from a walk of every
+ * contributing xattr and apply a corrective delta.  Should fsync()
+ * and avoid fsetattr() while operation is ongoing.
+ */
+struct scoutfs_ioctl_recompute_totl {
+	__u64	name[SCOUTFS_IOCTL_XATTR_TOTAL_NAME_NR];
+};
+
+#define SCOUTFS_IOC_RECOMPUTE_TOTL \
+	_IOW(SCOUTFS_IOCTL_MAGIC, 26, struct scoutfs_ioctl_recompute_totl)
+
 #endif

kmod/src/xattr.c

@@ -26,6 +26,9 @@
 #include "trans.h"
 #include "xattr.h"
 #include "lock.h"
+#include "totl.h"
+#include "wkic.h"
+#include "msg.h"
 #include "hash.h"
 #include "acl.h"
 #include "scoutfs_trace.h"
@@ -633,6 +636,220 @@ int scoutfs_xattr_combine_totl(void *dst, int dst_len, void *src, int src_len)
 	return SCOUTFS_DELTA_COMBINED;
 }
 
+struct recompute_totl_current {
+	u64 total;
+	u64 count;
+};
+
+static int recompute_totl_current_cb(struct scoutfs_key *key, void *val, unsigned int val_len,
+				     void *cb_arg)
+{
+	struct recompute_totl_current *cur = cb_arg;
+	struct scoutfs_xattr_totl_val *tval = val;
+
+	if (val_len != sizeof(*tval))
+		return -EIO;
+
+	cur->total = le64_to_cpu(tval->total);
+	cur->count = le64_to_cpu(tval->count);
+	return 0;
+}
+
+/*
+ * Recompute the totl item @target_name identifies by walking every
+ * xattr under per-inode-group READ locks, then apply a delta that
+ * aligns the wkic-read merged value with the recomputed sum.  Caller
+ * should fsync() and avoid fsetattr() while operation is ongoing.
+ */
+int scoutfs_xattr_recompute_totl(struct super_block *sb, u64 *target_name)
+{
+	struct scoutfs_xattr_prefix_tags tgs;
+	struct recompute_totl_current cur;
+	struct scoutfs_xattr_totl_val tval;
+	struct scoutfs_xattr *xat = NULL;
+	struct scoutfs_lock *tag_lock = NULL;
+	struct scoutfs_lock *lock = NULL;
+	struct scoutfs_key range_start;
+	struct scoutfs_key range_end;
+	struct scoutfs_key target_key;
+	struct scoutfs_key bounded_last;
+	struct scoutfs_key next_key;
+	struct scoutfs_key key;
+	struct scoutfs_key last;
+	unsigned int xat_bytes;
+	unsigned int val_len;
+	bool release = false;
+	s64 sum_total = 0;
+	s64 sum_count = 0;
+	s64 delta_total;
+	s64 delta_count;
+	void *value;
+	u64 trail[3];
+	u64 next_ino;
+	u64 total;
+	int ret;
+
+	xat_bytes = sizeof(struct scoutfs_xattr) + SCOUTFS_XATTR_MAX_NAME_LEN +
+		    SCOUTFS_XATTR_MAX_TOTL_U64;
+	xat = kmalloc(xat_bytes, GFP_NOFS);
+	if (!xat)
+		return -ENOMEM;
+
+	scoutfs_xattr_init_totl_key(&target_key, target_name);
+
+	init_xattr_key(&key, 0, 0, 0);
+	init_xattr_key(&last, U64_MAX, U32_MAX, U64_MAX);
+	last.skx_part = U8_MAX;
+
+	while (scoutfs_key_compare(&key, &last) <= 0) {
+		if (lock == NULL) {
+			ret = scoutfs_lock_ino(sb, SCOUTFS_LOCK_READ, 0,
+					       le64_to_cpu(key.skx_ino), &lock);
+			if (ret < 0)
+				goto out;
+		}
+
+		bounded_last = scoutfs_key_compare(&lock->end, &last) < 0 ? lock->end : last;
+
+		ret = scoutfs_item_next(sb, &key, &bounded_last, xat, xat_bytes, lock);
+		if (ret == -ENOENT) {
+			/* skip past empty regions, similar to scoutfs_ioc_walk_inodes() */
+			key = lock->end;
+			scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
+			lock = NULL;
+
+			scoutfs_key_inc(&key);
+			if (scoutfs_key_compare(&key, &last) > 0)
+				break;
+
+			ret = scoutfs_forest_next_hint(sb, &key, &next_key);
+			if (ret == -ENOENT || (ret == 0 &&
+			    scoutfs_key_compare(&next_key, &last) > 0)) {
+				ret = 0;
+				break;
+			}
+			if (ret < 0)
+				goto out;
+
+			key = next_key;
+			continue;
+		}
+		if (ret < 0)
+			goto out;
+
+		if (key.sk_type < SCOUTFS_XATTR_TYPE) {
+			init_xattr_key(&key, le64_to_cpu(key.skx_ino), 0, 0);
+			continue;
+		}
+		if (key.sk_type > SCOUTFS_XATTR_TYPE) {
+			next_ino = le64_to_cpu(key.skx_ino) + 1;
+			if (next_ino == 0)
+				break;
+			init_xattr_key(&key, next_ino, 0, 0);
+			continue;
+		}
+
+		if (key.skx_part != 0) {
+			scoutfs_key_inc(&key);
+			continue;
+		}
+
+		if (ret < sizeof(struct scoutfs_xattr) ||
+		    ret < offsetof(struct scoutfs_xattr, name[xat->name_len])) {
+			ret = -EIO;
+			goto out;
+		}
+
+		/* tag/name parse + value extract similar to scoutfs_xattr_drop_inode_items() */
+		if (scoutfs_xattr_parse_tags(xat->name, xat->name_len, &tgs) != 0 || !tgs.totl) {
+			scoutfs_key_inc(&key);
+			continue;
+		}
+
+		if (parse_dotted_u64s(trail, ARRAY_SIZE(trail), xat->name, xat->name_len) < 0) {
+			scoutfs_key_inc(&key);
+			continue;
+		}
+
+		if (trail[0] != target_name[0] || trail[1] != target_name[1] ||
+		    trail[2] != target_name[2]) {
+			scoutfs_key_inc(&key);
+			continue;
+		}
+
+		value = &xat->name[xat->name_len];
+		val_len = ret - offsetof(struct scoutfs_xattr, name[xat->name_len]);
+		if (val_len != le16_to_cpu(xat->val_len)) {
+			ret = -EIO;
+			goto out;
+		}
+
+		ret = parse_totl_u64(value, val_len, &total);
+		if (ret < 0)
+			goto out;
+
+		sum_total += (s64)total;
+		sum_count += 1;
+
+		scoutfs_key_inc(&key);
+	}
+
+	if (lock) {
+		scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
+		lock = NULL;
+	}
+
+	scoutfs_totl_set_range(&range_start, &range_end);
+	do {
+		memset(&cur, 0, sizeof(cur));
+		ret = scoutfs_wkic_iterate_stable(sb, &target_key, &target_key,
+						  &range_start, &range_end,
+						  recompute_totl_current_cb, &cur);
+	} while (ret == -ESTALE);
+	if (ret < 0)
+		goto out;
+
+	delta_total = sum_total - (s64)cur.total;
+	delta_count = sum_count - (s64)cur.count;
+
+	if (delta_total == 0 && delta_count == 0) {
+		ret = 0;
+		goto out;
+	}
+
+	scoutfs_warn(sb, "totl recompute applying delta total=%lld count=%lld to key "SK_FMT" (was total=%lld count=%lld, recomputed total=%lld count=%lld)",
+		     delta_total, delta_count, SK_ARG(&target_key),
+		     (s64)cur.total, (s64)cur.count, sum_total, sum_count);
+
+	tval.total = cpu_to_le64((u64)delta_total);
+	tval.count = cpu_to_le64((u64)delta_count);
+
+	/* same lock + trans + item_delta shape as scoutfs_ioc_inject_totl_delta() */
+	ret = scoutfs_lock_xattr_totl(sb, SCOUTFS_LOCK_WRITE_ONLY, 0, &tag_lock);
+	if (ret < 0)
+		goto out;
+
+	ret = scoutfs_hold_trans(sb, false);
+	if (ret < 0)
+		goto unlock_tag;
+	release = true;
+
+	ret = scoutfs_item_delta(sb, &target_key, &tval, sizeof(tval), tag_lock);
+
+	scoutfs_release_trans(sb);
+	release = false;
+
+unlock_tag:
+	scoutfs_unlock(sb, tag_lock, SCOUTFS_LOCK_WRITE_ONLY);
+out:
+	if (release)
+		scoutfs_release_trans(sb);
+	if (lock)
+		scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
+	kfree(xat);
+	return ret;
+}
+
 void scoutfs_xattr_indx_get_range(struct scoutfs_key *start, struct scoutfs_key *end)
 {
 	scoutfs_key_set_zeros(start);

kmod/src/xattr.h

@@ -30,6 +30,7 @@ int scoutfs_xattr_parse_tags(const char *name, unsigned int name_len,
 
 void scoutfs_xattr_init_totl_key(struct scoutfs_key *key, u64 *name);
 int scoutfs_xattr_combine_totl(void *dst, int dst_len, void *src, int src_len);
+int scoutfs_xattr_recompute_totl(struct super_block *sb, u64 *target_name);
 
 void scoutfs_xattr_indx_get_range(struct scoutfs_key *start, struct scoutfs_key *end);
 void scoutfs_xattr_init_indx_key(struct scoutfs_key *key, u8 major, u64 minor, u64 ino, u64 xid);