get-batch: amend abandoned work-item cleanup (part two)

* replace WI `demand` flag with `owned` ownership gate
* let asm() claim ownership before assembling and release it after cleanup
* let abandoned-WI GC clean only unowned work items
* store "WI abandonment" as an error; return it from asm()
* check WI terminal state while waiting on Rx (not only between asm() iters)
  - add inline helper to return error(nil)
* treat internal `errStopped` sentinel as quiet asm() termination
---
* part two, prev. commit: 926bdbb

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

xact/xs/lso.go

@@ -111,8 +111,8 @@ const (
 )
 
 var (
-	errStopped = errors.New("stopped")
-	ErrGone    = errors.New("gone")
+	errLsoStopped = errors.New("stopped")
+	ErrGone       = errors.New("gone")
 )
 
 // interface guard
@@ -822,7 +822,7 @@ func (r *LsoXact) doWalk(msg *apc.LsoMsg) {
 	}
 	opts.WalkOpts.Bck.Copy(r.Bck().Bucket())
 	if err := fs.WalkBck(opts); err != nil {
-		if err != filepath.SkipDir && err != errStopped {
+		if err != filepath.SkipDir && err != errLsoStopped {
 			r.AddErr(err, 0)
 		}
 	}
@@ -878,7 +878,7 @@ func (r *LsoXact) cb(fqn string, de fs.DirEntry) error {
 		select {
 		case r.walk.pageCh <- entry:
 		case <-r.walk.stopCh.Listen():
-			return errStopped
+			return errLsoStopped
 		}
 		return nil
 	}
@@ -897,7 +897,7 @@ func (r *LsoXact) cb(fqn string, de fs.DirEntry) error {
 	case r.walk.pageCh <- entry:
 		/* do nothing */
 	case <-r.walk.stopCh.Listen():
-		return errStopped
+		return errLsoStopped
 	}
 
 	if !msg.IsFlagSet(apc.LsArchDir) {
@@ -927,7 +927,7 @@ func (r *LsoXact) cb(fqn string, de fs.DirEntry) error {
 		case r.walk.pageCh <- e:
 			/* do nothing */
 		case <-r.walk.stopCh.Listen():
-			return errStopped
+			return errLsoStopped
 		}
 	}
 	return nil