Whamcloud - gitweb
LU-15886 lfsck: remove unreasonable assertions
[fs/lustre-release.git] / lustre / lfsck / lfsck_namespace.c
index 9606a11..df57d18 100644 (file)
@@ -87,7 +87,7 @@ static void lfsck_namespace_assistant_req_fini(const struct lu_env *env,
                                               struct lfsck_assistant_req *lar)
 {
        struct lfsck_namespace_req *lnr =
-                       container_of0(lar, struct lfsck_namespace_req, lnr_lar);
+               container_of(lar, struct lfsck_namespace_req, lnr_lar);
 
        if (lnr->lnr_lmv != NULL)
                lfsck_lmv_put(env, lnr->lnr_lmv);
@@ -531,7 +531,7 @@ int lfsck_namespace_trace_update(const struct lu_env *env,
                GOTO(log, rc);
        }
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -596,8 +596,7 @@ int lfsck_namespace_check_exist(const struct lu_env *env,
        if (unlikely(lfsck_is_dead_obj(obj)))
                RETURN(LFSCK_NAMEENTRY_DEAD);
 
-       rc = dt_lookup(env, dir, (struct dt_rec *)fid,
-                      (const struct dt_key *)name);
+       rc = dt_lookup_dir(env, dir, name, fid);
        if (rc == -ENOENT)
                RETURN(LFSCK_NAMEENTRY_REMOVED);
 
@@ -696,7 +695,7 @@ static int lfsck_namespace_links_remove(const struct lu_env *env,
 
        LASSERT(dt_object_remote(obj) == 0);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -831,7 +830,7 @@ static void lfsck_linkea_del_buf(struct linkea_data *ldata,
 
                ldata->ld_lee = NULL;
        } else {
-               linkea_del_buf(ldata, lname);
+               linkea_del_buf(ldata, lname, false);
        }
 }
 
@@ -946,8 +945,7 @@ again:
        do {
                namelen = snprintf(info->lti_key, NAME_MAX, DFID"%s-%s-%d",
                                   PFID(cfid), infix, type, idx++);
-               rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                              (const struct dt_key *)info->lti_key);
+               rc = dt_lookup_dir(env, parent, info->lti_key, &tfid);
                if (rc != 0 && rc != -ENOENT)
                        GOTO(log, rc);
 
@@ -962,8 +960,7 @@ again:
 
        /* Re-check whether the name conflict with othrs after taken
         * the ldlm lock. */
-       rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                      (const struct dt_key *)info->lti_key);
+       rc = dt_lookup_dir(env, parent, info->lti_key, &tfid);
        if (rc == 0) {
                if (!lu_fid_eq(cfid, &tfid)) {
                        exist = false;
@@ -993,7 +990,7 @@ again:
 
        lfsck_buf_init(&linkea_buf, ldata2.ld_buf->lb_buf,
                       ldata2.ld_leh->leh_len);
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -1125,11 +1122,11 @@ static int lfsck_lmv_set(const struct lu_env *env,
 
        ENTRY;
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
 
-       rc = dt_declare_xattr_set(env, obj, &buf, XATTR_NAME_LMV".set", 0, th);
+       rc = dt_declare_xattr_set(env, obj, &buf, XATTR_NAME_LMV, 0, th);
        if (rc)
                GOTO(stop, rc);
 
@@ -1137,7 +1134,7 @@ static int lfsck_lmv_set(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
-       rc = dt_xattr_set(env, obj, &buf, XATTR_NAME_LMV".set", 0, th);
+       rc = dt_xattr_set(env, obj, &buf, XATTR_NAME_LMV, 0, th);
        if (rc)
                GOTO(stop, rc);
 
@@ -1158,7 +1155,7 @@ static int lfsck_lmv_delete(const struct lu_env *env,
 
        ENTRY;
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
 
@@ -1323,7 +1320,7 @@ static int lfsck_namespace_insert_normal(const struct lu_env *env,
        if (unlikely(!dt_try_as_dir(env, parent)))
                GOTO(unlock, rc = -ENOTDIR);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 
@@ -1341,8 +1338,8 @@ static int lfsck_namespace_insert_normal(const struct lu_env *env,
        }
 
        if (parent_lmv_lost) {
-               rc = dt_declare_xattr_set(env, parent, &buf,
-                                         XATTR_NAME_LMV".set", 0, th);
+               rc = dt_declare_xattr_set(env, parent, &buf, XATTR_NAME_LMV,
+                                         0, th);
                if (rc)
                        GOTO(stop, rc);
        }
@@ -1375,8 +1372,7 @@ static int lfsck_namespace_insert_normal(const struct lu_env *env,
        }
 
        if (parent_lmv_lost) {
-               rc = dt_xattr_set(env, parent, &buf, XATTR_NAME_LMV".set", 0,
-                                 th);
+               rc = dt_xattr_set(env, parent, &buf, XATTR_NAME_LMV, 0, th);
                if (rc)
                        GOTO(stop, rc);
        }
@@ -1468,16 +1464,15 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
                GOTO(log, rc = 1);
 
        if (dt_object_remote(orphan)) {
-               LASSERT(lfsck->li_lpf_root_obj != NULL);
+               if (lfsck->li_lpf_root_obj == NULL)
+                       GOTO(log, rc = -EBADF);
 
                idx = lfsck_find_mdt_idx_by_fid(env, lfsck, cfid);
                if (idx < 0)
                        GOTO(log, rc = idx);
 
                snprintf(name, 8, "MDT%04x", idx);
-               rc = dt_lookup(env, lfsck->li_lpf_root_obj,
-                              (struct dt_rec *)&tfid,
-                              (const struct dt_key *)name);
+               rc = dt_lookup_dir(env, lfsck->li_lpf_root_obj, name, &tfid);
                if (rc != 0)
                        GOTO(log, rc = (rc == -ENOENT ? -ENXIO : rc));
 
@@ -1504,8 +1499,7 @@ again:
        do {
                namelen = snprintf(name, 31, DFID"-P-%d",
                                   PFID(cfid), idx++);
-               rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                              (const struct dt_key *)name);
+               rc = dt_lookup_dir(env, parent, name, &tfid);
                if (rc != 0 && rc != -ENOENT)
                        GOTO(log, rc);
        } while (rc == 0);
@@ -1517,8 +1511,7 @@ again:
 
        /* Re-check whether the name conflict with othrs after taken
         * the ldlm lock. */
-       rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                      (const struct dt_key *)name);
+       rc = dt_lookup_dir(env, parent, name, &tfid);
        if (unlikely(rc == 0)) {
                lfsck_unlock(llh);
                goto again;
@@ -1546,7 +1539,7 @@ again:
        if (rc != 0)
                GOTO(unlock1, rc);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(unlock1, rc = PTR_ERR(th));
 
@@ -1584,8 +1577,8 @@ again:
                lmv->lmv_master_mdt_index = lfsck_dev_idx(lfsck);
                lfsck_lmv_header_cpu_to_le(lmv2, lmv);
                lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));
-               rc = dt_declare_xattr_set(env, orphan, &lmv_buf,
-                                         XATTR_NAME_LMV, 0, th);
+               rc = dt_declare_xattr_set(env, orphan, &lmv_buf, XATTR_NAME_LMV,
+                                         0, th);
                if (rc != 0)
                        GOTO(stop, rc);
        }
@@ -1738,7 +1731,7 @@ static int lfsck_namespace_shrink_linkea(const struct lu_env *env,
        }
 
 again:
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(unlock1, rc = PTR_ERR(th));
 
@@ -1990,8 +1983,7 @@ static int lfsck_namespace_replace_cond(const struct lu_env *env,
                goto replace;
        }
 
-       rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                      (const struct dt_key *)name);
+       rc = dt_lookup_dir(env, parent, name, &tfid);
        if (rc == -ENOENT) {
                exist = false;
                goto replace;
@@ -2054,7 +2046,7 @@ replace:
        if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
                GOTO(log, rc = 1);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -2140,7 +2132,7 @@ int lfsck_namespace_rebuild_linkea(const struct lu_env *env,
        int                              rc     = 0;
        ENTRY;
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -2246,7 +2238,7 @@ int lfsck_namespace_repair_dirent(const struct lu_env *env,
        if (rc != 0)
                GOTO(log, rc);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(unlock1, rc = PTR_ERR(th));
 
@@ -2277,8 +2269,7 @@ int lfsck_namespace_repair_dirent(const struct lu_env *env,
 
 
        dt_write_lock(env, parent, 0);
-       rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                      (const struct dt_key *)name);
+       rc = dt_lookup_dir(env, dt_object_child(parent), name, &tfid);
        /* Someone has removed the bad name entry by race. */
        if (rc == -ENOENT)
                GOTO(unlock2, rc = 0);
@@ -2394,7 +2385,7 @@ static int lfsck_namespace_repair_unmatched_pairs(const struct lu_env *env,
        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -2676,8 +2667,7 @@ lost_parent:
                GOTO(out, rc);
        }
 
-       rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                      (const struct dt_key *)cname->ln_name);
+       rc = dt_lookup_dir(env, parent, cname->ln_name, &tfid);
        if (rc == -ENOENT) {
                /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT
                 * has ever tried to verify some remote MDT-object that resides
@@ -2893,8 +2883,20 @@ again:
                }
 
                parent = lfsck_object_find_bottom(env, lfsck, &tfid);
-               if (IS_ERR(parent))
-                       RETURN(PTR_ERR(parent));
+               if (IS_ERR(parent)) {
+                       rc = PTR_ERR(parent);
+                       /* if @pfid doesn't have a valid OI mapping, it will
+                        * trigger OI scrub, and -ENONET is is returned if it's
+                        * remote, -EINPROGRESS if local.
+                        */
+                       if ((rc == -ENOENT || rc == -EINPROGRESS) &&
+                           ldata->ld_leh->leh_reccount > 1) {
+                               lfsck_linkea_del_buf(ldata, cname);
+                               continue;
+                       }
+
+                       RETURN(rc);
+               }
 
                if (!dt_object_exists(parent)) {
                        lfsck_object_put(env, parent);
@@ -2917,8 +2919,7 @@ again:
                        continue;
                }
 
-               rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
-                              (const struct dt_key *)cname->ln_name);
+               rc = dt_lookup_dir(env, parent, cname->ln_name, &tfid);
                *pfid2 = *lfsck_dto2fid(parent);
                if (rc == -ENOENT) {
                        lfsck_object_put(env, parent);
@@ -3107,7 +3108,7 @@ static int lfsck_namespace_repair_nlink(const struct lu_env *env,
        if (rc != 0)
                GOTO(log, rc);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -3248,11 +3249,8 @@ static int lfsck_namespace_double_scan_dir(const struct lu_env *env,
 
        if (flags & (LNTF_CHECK_LINKEA | LNTF_CHECK_PARENT) &&
            !(lfsck->li_bookmark_ram.lb_param & LPF_ALL_TGT)) {
-               CDEBUG(D_LFSCK, "%s: some MDT(s) maybe NOT take part in the"
-                      "the namespace LFSCK, then the LFSCK cannot guarantee"
-                      "all the name entries have been verified in first-stage"
-                      "scanning. So have to skip orphan related handling for"
-                      "the directory object "DFID" with remote name entry\n",
+               CDEBUG(D_LFSCK,
+                      "%s: some MDT(s) maybe NOT take part in the the namespace LFSCK, then the LFSCK cannot guarantee all the name entries have been verified in first-stage scanning. So have to skip orphan related handling for the directory object "DFID" with remote name entry\n",
                       lfsck_lfsck2name(lfsck), PFID(cfid));
 
                RETURN(0);
@@ -3280,8 +3278,7 @@ lock:
                GOTO(out, rc = 0);
        }
 
-       rc = dt_lookup(env, child, (struct dt_rec *)pfid,
-                      (const struct dt_key *)dotdot);
+       rc = dt_lookup_dir(env, child, dotdot, pfid);
        if (rc != 0) {
                if (rc != -ENOENT && rc != -ENODATA && rc != -EINVAL) {
                        dt_read_unlock(env, child);
@@ -3471,7 +3468,7 @@ static int lfsck_namespace_linkea_clear_overflow(const struct lu_env *env,
        if (rc != 0)
                GOTO(log, rc);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(log, rc = PTR_ERR(th));
 
@@ -3497,10 +3494,6 @@ static int lfsck_namespace_linkea_clear_overflow(const struct lu_env *env,
        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);
@@ -3610,7 +3603,7 @@ static int lfsck_namespace_check_agent_entry(const struct lu_env *env,
                if (rc)
                        GOTO(out, rc);
 
-               handle = dt_trans_create(env, dev);
+               handle = lfsck_trans_create(env, dev, lfsck);
                if (IS_ERR(handle))
                        GOTO(unlock, rc = PTR_ERR(handle));
 
@@ -3782,12 +3775,22 @@ static int lfsck_namespace_double_scan_one(const struct lu_env *env,
 
                        repaired = true;
 
-                       /* fall through */
+                       /* fallthrough */
                }
 
                parent = lfsck_object_find_bottom(env, lfsck, pfid);
-               if (IS_ERR(parent))
-                       GOTO(out, rc = PTR_ERR(parent));
+               if (IS_ERR(parent)) {
+                       rc = PTR_ERR(parent);
+                       /* if @pfid doesn't have a valid OI mapping, it will
+                        * trigger OI scrub, and -ENONET is is returned if it's
+                        * remote, -EINPROGRESS if local.
+                        */
+                       if ((rc == -ENOENT || rc == -EINPROGRESS) &&
+                           ldata.ld_leh->leh_reccount > 1)
+                               rc = lfsck_namespace_shrink_linkea(env, com,
+                                       child, &ldata, cname, pfid, true);
+                       GOTO(out, rc);
+               }
 
                if (!dt_object_exists(parent)) {
 
@@ -3871,8 +3874,7 @@ lost_parent:
                        continue;
                }
 
-               rc = dt_lookup(env, parent, (struct dt_rec *)cfid,
-                              (const struct dt_key *)cname->ln_name);
+               rc = dt_lookup_dir(env, parent, cname->ln_name, cfid);
                if (rc != 0 && rc != -ENOENT) {
                        lfsck_object_put(env, parent);
 
@@ -4361,7 +4363,7 @@ static void lfsck_namespace_close_dir(const struct lu_env *env,
        lad->lad_prefetched++;
        spin_unlock(&lad->lad_lock);
        if (wakeup)
-               wake_up_all(&lad->lad_thread.t_ctl_waitq);
+               wake_up(&lad->lad_thread.t_ctl_waitq);
 
        EXIT;
 }
@@ -4538,20 +4540,32 @@ static int lfsck_namespace_exec_oit(const struct lu_env *env,
                                    struct lfsck_component *com,
                                    struct dt_object *obj)
 {
-       struct lfsck_thread_info *info  = lfsck_env_info(env);
-       struct lfsck_namespace   *ns    = com->lc_file_ram;
-       struct lfsck_instance    *lfsck = com->lc_lfsck;
-       const struct lu_fid      *fid   = lfsck_dto2fid(obj);
-       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);
-       struct linkea_data        ldata = { NULL };
-       __u32                     idx   = lfsck_dev_idx(lfsck);
-       int                       rc;
+       struct lfsck_thread_info *info = lfsck_env_info(env);
+       struct lfsck_namespace *ns = com->lc_file_ram;
+       struct lfsck_instance *lfsck = com->lc_lfsck;
+       const struct lu_fid *fid = lfsck_dto2fid(obj);
+       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);
+       struct linkea_data ldata = { NULL };
+       __u32 idx = lfsck_dev_idx(lfsck);
+       struct lu_attr la = { .la_valid = 0 };
        bool remote = false;
+       int rc;
        ENTRY;
 
+       rc = dt_attr_get(env, obj, &la);
+       if (unlikely(rc || (la.la_valid & LA_FLAGS &&
+                           la.la_flags & LUSTRE_ORPHAN_FL))) {
+               CDEBUG(D_INFO,
+                      "%s: skip orphan "DFID", %llx/%x: rc = %d\n",
+                      lfsck_lfsck2name(lfsck), PFID(fid),
+                      la.la_valid, la.la_flags, rc);
+
+               return rc;
+       }
+
        rc = lfsck_links_read(env, obj, &ldata);
        if (rc == -ENOENT)
                GOTO(out, rc = 0);
@@ -4656,14 +4670,12 @@ static int lfsck_namespace_exec_dir(const struct lu_env *env,
        struct lfsck_bookmark           *bk      = &lfsck->li_bookmark_ram;
        struct ptlrpc_thread            *mthread = &lfsck->li_thread;
        struct ptlrpc_thread            *athread = &lad->lad_thread;
-       struct l_wait_info               lwi     = { 0 };
        bool                             wakeup  = false;
 
-       l_wait_event(mthread->t_ctl_waitq,
-                    lad->lad_prefetched < bk->lb_async_windows ||
-                    !thread_is_running(mthread) ||
-                    !thread_is_running(athread),
-                    &lwi);
+       wait_event_idle(mthread->t_ctl_waitq,
+                       lad->lad_prefetched < bk->lb_async_windows ||
+                       !thread_is_running(mthread) ||
+                       !thread_is_running(athread));
 
        if (unlikely(!thread_is_running(mthread) ||
                     !thread_is_running(athread)))
@@ -4696,7 +4708,7 @@ static int lfsck_namespace_exec_dir(const struct lu_env *env,
        lad->lad_prefetched++;
        spin_unlock(&lad->lad_lock);
        if (wakeup)
-               wake_up_all(&lad->lad_thread.t_ctl_waitq);
+               wake_up(&lad->lad_thread.t_ctl_waitq);
 
        down_write(&com->lc_sem);
        com->lc_new_checked++;
@@ -5138,7 +5150,7 @@ static int lfsck_namespace_in_notify(const struct lu_env *env,
                stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
                lfsck_stop(env, lfsck->li_bottom, stop);
        } else if (lfsck_phase2_next_ready(lad)) {
-               wake_up_all(&lad->lad_thread.t_ctl_waitq);
+               wake_up(&lad->lad_thread.t_ctl_waitq);
        }
 
        RETURN(0);
@@ -5221,7 +5233,7 @@ static int lfsck_namespace_query(const struct lu_env *env,
        return rc;
 }
 
-static struct lfsck_operations lfsck_namespace_ops = {
+static const struct lfsck_operations lfsck_namespace_ops = {
        .lfsck_reset            = lfsck_namespace_reset,
        .lfsck_fail             = lfsck_namespace_fail,
        .lfsck_close_dir        = lfsck_namespace_close_dir,
@@ -5364,6 +5376,8 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
         */
        if (S_ISREG(type))
                child->do_ops->do_ah_init(env, hint,  parent, child, type);
+       else if (S_ISDIR(type))
+               child->do_ops->do_ah_init(env, hint,  NULL, child, type);
 
        memset(dof, 0, sizeof(*dof));
        dof->dof_type = dt_mode_to_dft(type);
@@ -5371,7 +5385,7 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
         * the MDT-object without stripes (dof->dof_reg.striped = 0). related
         * OST-objects will be created when write open. */
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(unlock_child, rc = PTR_ERR(th));
 
@@ -5423,7 +5437,7 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
                        lfsck_lmv_header_cpu_to_le(lmv2, lmv2);
                        lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));
                        rc = dt_declare_xattr_set(env, child, &lmv_buf,
-                                                 XATTR_NAME_LMV".set", 0, th);
+                                                 XATTR_NAME_LMV, 0, th);
                        if (rc != 0)
                                GOTO(stop, rc);
                }
@@ -5485,8 +5499,8 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
 
                /* 5b. generate slave LMV EA. */
                if (lnr->lnr_lmv != NULL && lnr->lnr_lmv->ll_lmv_master) {
-                       rc = dt_xattr_set(env, child, &lmv_buf,
-                                         XATTR_NAME_LMV".set", 0, th);
+                       rc = dt_xattr_set(env, child, &lmv_buf, XATTR_NAME_LMV,
+                                         0, th);
                        if (rc != 0)
                                GOTO(unlock, rc);
                }
@@ -5558,7 +5572,7 @@ static int lfsck_namespace_assistant_handler_p1(const struct lu_env *env,
        const struct lu_name       *cname;
        struct thandle             *handle   = NULL;
        struct lfsck_namespace_req *lnr      =
-                       container_of0(lar, struct lfsck_namespace_req, lnr_lar);
+               container_of(lar, struct lfsck_namespace_req, lnr_lar);
        struct dt_object           *dir      = NULL;
        struct dt_object           *obj      = NULL;
        struct lfsck_assistant_object *lso   = lar->lar_parent;
@@ -5567,8 +5581,8 @@ static int lfsck_namespace_assistant_handler_p1(const struct lu_env *env,
        struct lustre_handle        lh       = { 0 };
        bool                        repaired = false;
        bool                        dtlocked = false;
-       bool                        remove;
-       bool                        newdata;
+       bool                        remove = false;
+       bool                        newdata = false;
        bool                        log      = false;
        bool                        bad_hash = false;
        bool                        bad_linkea = false;
@@ -5592,11 +5606,9 @@ static int lfsck_namespace_assistant_handler_p1(const struct lu_env *env,
                repaired = true;
        }
 
-       if (unlikely(fid_is_zero(&lnr->lnr_fid))) {
-               if (strcmp(lnr->lnr_name, dotdot) != 0)
-                       LBUG();
-               else
-                       rc = lfsck_namespace_trace_update(env, com, pfid,
+       if (unlikely(fid_is_zero(&lnr->lnr_fid) &&
+                    strcmp(lnr->lnr_name, dotdot) == 0)) {
+               rc = lfsck_namespace_trace_update(env, com, pfid,
                                                LNTF_CHECK_PARENT, true);
 
                GOTO(out, rc);
@@ -5715,7 +5727,7 @@ again:
                if (rc != 0)
                        GOTO(out, rc);
 
-               handle = dt_trans_create(env, dev);
+               handle = lfsck_trans_create(env, dev, lfsck);
                if (IS_ERR(handle))
                        GOTO(out, rc = PTR_ERR(handle));
 
@@ -5813,6 +5825,17 @@ again:
                newdata = true;
 
 nodata:
+               if (rc == -ENOENT &&
+                   linkea_will_overflow(&ldata, cname)) {
+                       CDEBUG(D_INODE, "No enough space to hold linkea entry '"
+                              DFID": %.*s' at %u\n", PFID(pfid),
+                              cname->ln_namelen, cname->ln_name,
+                              ldata.ld_leh->leh_overflow_time);
+                       log = true;
+                       rc = 0;
+                       goto stop;
+               }
+
                if (bk->lb_param & LPF_DRYRUN) {
                        if (rc == -ENODATA)
                                ns->ln_flags |= LF_UPGRADE;
@@ -5868,7 +5891,7 @@ nodata:
                                GOTO(stop, rc);
                }
 
-               rc = linkea_add_buf(&ldata, cname, pfid);
+               rc = linkea_add_buf(&ldata, cname, pfid, false);
                if (rc == 0)
                        rc = lfsck_links_write(env, obj, &ldata, handle);
                if (rc != 0)
@@ -5955,6 +5978,17 @@ out:
                if (obj != NULL && count == 1 &&
                    S_ISREG(lfsck_object_type(obj)))
                        dt_attr_get(env, obj, la);
+
+               /* if new linkea entry is added, the old entry may be stale,
+                * check it in phase 2. Sigh, linkea check can only be done
+                * locally.
+                */
+               if (bad_linkea && !remove && !newdata &&
+                   !dt_object_remote(obj) && count > 1)
+                       rc = lfsck_namespace_trace_update(env, com,
+                                                         &lnr->lnr_fid,
+                                                         LNTF_CHECK_LINKEA,
+                                                         true);
        }
 
 trace:
@@ -6092,13 +6126,19 @@ static int lfsck_namespace_scan_local_lpf_one(const struct lu_env *env,
        int                              rc     = 0;
        __u8                             flags  = 0;
        bool                             exist  = false;
+
        ENTRY;
 
        child = lfsck_object_find_by_dev(env, dev, &ent->lde_fid);
        if (IS_ERR(child))
                RETURN(PTR_ERR(child));
 
-       LASSERT(dt_object_exists(child));
+       if (!dt_object_exists(child)) {
+               CDEBUG(D_LFSCK, "%s: lost+found/%s doesn't exist\n",
+                      lfsck_lfsck2name(lfsck), ent->lde_name);
+               GOTO(out, rc = -ENOENT);
+       }
+
        LASSERT(!dt_object_remote(child));
 
        idx = lfsck_sub_trace_file_fid2idx(&ent->lde_fid);
@@ -6115,7 +6155,7 @@ static int lfsck_namespace_scan_local_lpf_one(const struct lu_env *env,
                GOTO(out, rc);
        }
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(out, rc = PTR_ERR(th));
 
@@ -6822,7 +6862,7 @@ out:
        EXIT;
 }
 
-struct lfsck_assistant_operations lfsck_namespace_assistant_ops = {
+const struct lfsck_assistant_operations lfsck_namespace_assistant_ops = {
        .la_handler_p1          = lfsck_namespace_assistant_handler_p1,
        .la_handler_p2          = lfsck_namespace_assistant_handler_p2,
        .la_fill_pos            = lfsck_namespace_assistant_fill_pos,
@@ -6844,8 +6884,9 @@ struct lfsck_assistant_operations lfsck_namespace_assistant_ops = {
  * \retval             0 for success
  * \retval             negative error number on failure
  */
-int lfsck_verify_linkea(const struct lu_env *env, struct dt_object *obj,
-                       const struct lu_name *cname, const struct lu_fid *pfid)
+int lfsck_verify_linkea(const struct lu_env *env, struct lfsck_instance *lfsck,
+                       struct dt_object *obj, const struct lu_name *cname,
+                       const struct lu_fid *pfid)
 {
        struct dt_device        *dev    = lfsck_obj2dev(obj);
        struct linkea_data       ldata  = { NULL };
@@ -6854,9 +6895,14 @@ int lfsck_verify_linkea(const struct lu_env *env, struct dt_object *obj,
        int                      rc;
        int                      fl     = LU_XATTR_CREATE;
        bool                     dirty  = false;
+
        ENTRY;
 
-       LASSERT(S_ISDIR(lfsck_object_type(obj)));
+       if (!dt_object_exists(obj))
+               RETURN(-ENOENT);
+
+       if (!S_ISDIR(lfsck_object_type(obj)))
+               RETURN(-ENOTDIR);
 
        rc = lfsck_links_read_with_rec(env, obj, &ldata);
        if (rc == -ENODATA) {
@@ -6882,7 +6928,7 @@ int lfsck_verify_linkea(const struct lu_env *env, struct dt_object *obj,
 
        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
 
@@ -6976,7 +7022,7 @@ int lfsck_update_name_entry(const struct lu_env *env,
        if (rc != 0)
                RETURN(rc);
 
-       th = dt_trans_create(env, dev);
+       th = lfsck_trans_create(env, dev, lfsck);
        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 
@@ -7114,14 +7160,14 @@ int lfsck_namespace_setup(const struct lu_env *env,
        default:
                CERROR("%s: unknown lfsck_namespace status %d\n",
                       lfsck_lfsck2name(lfsck), ns->ln_status);
-               /* fall through */
+               fallthrough;
        case LS_SCANNING_PHASE1:
        case LS_SCANNING_PHASE2:
                /* No need to store the status to disk right now.
                 * If the system crashed before the status stored,
                 * it will be loaded back when next time. */
                ns->ln_status = LS_CRASHED;
-               /* fall through */
+               fallthrough;
        case LS_PAUSED:
        case LS_CRASHED:
                spin_lock(&lfsck->li_lock);