+static inline bool
+lfsck_namespace_linkea_stale_overflow(struct linkea_data *ldata,
+ struct lfsck_namespace *ns)
+{
+ /* Both the leh_overflow_time and ln_time_latest_reset are
+ * local time based, so need NOT to care about clock drift
+ * among the servers. */
+ return ldata->ld_leh->leh_overflow_time &&
+ ldata->ld_leh->leh_overflow_time < ns->ln_time_latest_reset;
+}
+
+/**
+ * Clear the object's linkEA overflow timestamp.
+ *
+ * If the MDT-object has too many hard links as to the linkEA cannot hold
+ * all of them, then overflow timestamp will be set in the linkEA header.
+ * If some hard links are removed after that, then it is possible to hold
+ * other missed linkEA entries. If the namespace LFSCK have added all the
+ * related linkEA entries, then it will remove the overflow timestamp.
+ *
+ * \param[in] env pointer to the thread context
+ * \param[in] com pointer to the lfsck component
+ * \param[in] ldata pointer to the linkEA data for the given @obj
+ * \param[in] obj pointer to the dt_object to be handled
+ *
+ * \retval positive number for repaired cases
+ * \retval 0 if nothing to be repaired
+ * \retval negative error number on failure
+ */
+static int lfsck_namespace_linkea_clear_overflow(const struct lu_env *env,
+ struct lfsck_component *com,
+ struct linkea_data *ldata,
+ struct dt_object *obj)
+{
+ struct lfsck_namespace *ns = com->lc_file_ram;
+ struct lfsck_instance *lfsck = com->lc_lfsck;
+ struct dt_device *dev = lfsck_obj2dev(obj);
+ struct thandle *th = NULL;
+ struct lustre_handle lh = { 0 };
+ struct lu_buf linkea_buf;
+ int rc = 0;
+ ENTRY;
+
+ LASSERT(!dt_object_remote(obj));
+
+ rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
+ MDS_INODELOCK_UPDATE, LCK_PW);
+ if (rc != 0)
+ GOTO(log, rc);
+
+ th = dt_trans_create(env, dev);
+ if (IS_ERR(th))
+ GOTO(log, rc = PTR_ERR(th));
+
+ rc = dt_declare_xattr_set(env, obj,
+ lfsck_buf_get_const(env, NULL, MAX_LINKEA_SIZE),
+ XATTR_NAME_LINK, 0, th);
+ if (rc != 0)
+ GOTO(stop, rc);
+
+ rc = dt_trans_start_local(env, dev, th);
+ if (rc != 0)
+ GOTO(stop, rc);
+
+ dt_write_lock(env, obj, 0);
+ rc = lfsck_links_read(env, obj, ldata);
+ if (rc != 0)
+ GOTO(unlock, rc);
+
+ if (unlikely(!lfsck_namespace_linkea_stale_overflow(ldata, ns)))
+ GOTO(unlock, rc = 0);
+
+ ldata->ld_leh->leh_overflow_time = 0;
+ if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
+ GOTO(unlock, rc = 1);
+
+ /* If all known entries are in the linkEA, then the 'leh_reccount'
+ * should NOT be zero. */
+ LASSERT(ldata->ld_leh->leh_reccount > 0);
+
+ lfsck_buf_init(&linkea_buf, ldata->ld_buf->lb_buf,
+ ldata->ld_leh->leh_len);
+ rc = dt_xattr_set(env, obj, &linkea_buf, XATTR_NAME_LINK, 0, th);
+ if (unlikely(rc == -ENOSPC))
+ rc = 0;
+ else if (!rc)
+ rc = 1;
+
+ GOTO(unlock, rc);
+
+unlock:
+ dt_write_unlock(env, obj);
+
+stop:
+ dt_trans_stop(env, dev, th);
+
+log:
+ lfsck_ibits_unlock(&lh, LCK_PW);
+ CDEBUG(D_LFSCK, "%s: clear linkea overflow timestamp for the object "
+ DFID": rc = %d\n",
+ lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
+
+ return rc;
+}
+
+/**
+ * Verify the object's agent entry.
+ *
+ * If the object claims to have agent entry but the linkEA does not contain
+ * remote parent, then remove the agent entry. Otherwise, if the object has
+ * no agent entry but its linkEA contains remote parent, then will generate
+ * agent entry for it.
+ *
+ * \param[in] env pointer to the thread context
+ * \param[in] com pointer to the lfsck component
+ * \param[in] obj pointer to the dt_object to be handled
+ *
+ * \retval positive number for repaired cases
+ * \retval 0 if nothing to be repaired
+ * \retval negative error number on failure
+ */
+static int lfsck_namespace_check_agent_entry(const struct lu_env *env,
+ struct lfsck_component *com,
+ struct dt_object *obj)
+{
+ struct linkea_data ldata = { NULL };
+ struct lfsck_thread_info *info = lfsck_env_info(env);
+ struct lfsck_namespace *ns = com->lc_file_ram;
+ struct lfsck_instance *lfsck = com->lc_lfsck;
+ struct lu_fid *pfid = &info->lti_fid2;
+ struct lu_name *cname = &info->lti_name;
+ struct lu_seq_range *range = &info->lti_range;
+ struct seq_server_site *ss = lfsck_dev_site(lfsck);
+ __u32 idx = lfsck_dev_idx(lfsck);
+ int rc;
+ bool remote = false;
+ ENTRY;
+
+ if (!(lfsck->li_bookmark_ram.lb_param & LPF_ALL_TGT))
+ RETURN(0);
+
+ rc = lfsck_links_read_with_rec(env, obj, &ldata);
+ if (rc == -ENOENT || rc == -ENODATA)
+ RETURN(0);
+
+ if (rc && rc != -EINVAL)
+ GOTO(out, rc);
+
+ /* We check the agent entry again after verifying the linkEA
+ * successfully. So invalid linkEA should be dryrun mode. */
+ if (rc == -EINVAL || unlikely(!ldata.ld_leh->leh_reccount))
+ RETURN(0);
+
+ linkea_first_entry(&ldata);
+ while (ldata.ld_lee != NULL && !remote) {
+ linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen,
+ cname, pfid);
+ if (!linkea_entry_is_valid(&ldata, cname, pfid))
+ GOTO(out, rc = 0);
+
+ fld_range_set_mdt(range);
+ rc = fld_server_lookup(env, ss->ss_server_fld,
+ fid_seq(pfid), range);
+ if (rc)
+ GOTO(out, rc = (rc == -ENOENT ? 0 : rc));
+
+ if (range->lsr_index != idx)
+ remote = true;
+ else
+ linkea_next_entry(&ldata);
+ }
+
+ if ((lu_object_has_agent_entry(&obj->do_lu) && !remote) ||
+ (!lu_object_has_agent_entry(&obj->do_lu) && remote)) {
+ struct dt_device *dev = lfsck_obj2dev(obj);
+ struct linkea_data ldata2 = { NULL };
+ struct lustre_handle lh = { 0 };
+ struct lu_buf linkea_buf;
+ struct thandle *handle;
+
+ if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
+ GOTO(out, rc = 1);
+
+ rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
+ MDS_INODELOCK_UPDATE |
+ MDS_INODELOCK_XATTR, LCK_EX);
+ if (rc)
+ GOTO(out, rc);
+
+ handle = dt_trans_create(env, dev);
+ if (IS_ERR(handle))
+ GOTO(unlock, rc = PTR_ERR(handle));
+
+ lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
+ ldata.ld_leh->leh_len);
+ rc = dt_declare_xattr_set(env, obj, &linkea_buf,
+ XATTR_NAME_LINK, LU_XATTR_REPLACE, handle);
+ if (rc)
+ GOTO(stop, rc);
+
+ rc = dt_trans_start_local(env, dev, handle);
+ if (rc)
+ GOTO(stop, rc);
+
+ dt_write_lock(env, obj, 0);
+ rc = lfsck_links_read2_with_rec(env, obj, &ldata2);
+ if (rc) {
+ if (rc == -ENOENT || rc == -ENODATA)
+ rc = 0;
+ GOTO(unlock2, rc);
+ }
+
+ /* If someone changed linkEA by race, then the agent
+ * entry will be updated by lower layer automatically. */
+ if (ldata.ld_leh->leh_len != ldata2.ld_leh->leh_len ||
+ memcmp(ldata.ld_buf->lb_buf, ldata2.ld_buf->lb_buf,
+ ldata.ld_leh->leh_len) != 0)
+ GOTO(unlock2, rc = 0);
+
+ rc = dt_xattr_set(env, obj, &linkea_buf, XATTR_NAME_LINK,
+ LU_XATTR_REPLACE, handle);
+ if (!rc)
+ rc = 1;
+
+ GOTO(unlock2, rc);
+
+unlock2:
+ dt_write_unlock(env, obj);
+stop:
+ dt_trans_stop(env, dev, handle);
+unlock:
+ lfsck_ibits_unlock(&lh, LCK_EX);
+ }
+
+ GOTO(out, rc);
+
+out:
+ if (rc > 0)
+ ns->ln_agent_entries_repaired++;
+ if (rc)
+ CDEBUG(D_LFSCK, "%s: repair agent entry for "DFID": rc = %d\n",
+ lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
+ return rc;
+}
+