Sync agent queue replacements before success

[onechat] / tree.go

diff --git a/tree.go b/tree.go
index 4309597387a01ff0b6fbfc8e697a7c4a82f84830..765016eaba3e14c1ebd2dba4b5b30356346418e4 100644 (file)
--- a/tree.go
+++ b/tree.go
@@ -93,19 +93,19 @@ func (t *Tree) open(ctx context.Context, total int) error {
 
        t.order, err = openRecords(filepath.Join(t.dir, "tree.idx"), treeRecordBytes)
        if err != nil {
 
        t.order, err = openRecords(filepath.Join(t.dir, "tree.idx"), treeRecordBytes)
        if err != nil {

-               return err
+               return fmt.Errorf("open node order index: %w", err)

        }
 
        start := Position{}
        if t.order.count == 0 {
                err = t.removeLevels()
                if err != nil {
        }
 
        start := Position{}
        if t.order.count == 0 {
                err = t.removeLevels()
                if err != nil {

-                       return err
+                       return fmt.Errorf("remove empty level indexes: %w", err)

                }
        } else {
                record, _, err := t.order.Read(t.order.count - 1)
                if err != nil {
                }
        } else {
                record, _, err := t.order.Read(t.order.count - 1)
                if err != nil {

-                       return err
+                       return fmt.Errorf("read last node order: %w", err)

                }
 
                // A crash can stop an add between the two index writes, so the
                }
 
                // A crash can stop an add between the two index writes, so the


@@ -113,7 +113,7 @@ func (t *Tree) open(ctx context.Context, total int) error {
                node, last := treeDecodeRecord(record)
                err = t.writeLevel(node, last)
                if err != nil {
                node, last := treeDecodeRecord(record)
                err = t.writeLevel(node, last)
                if err != nil {

-                       return err
+                       return fmt.Errorf("restore last node level: %w", err)

                }
 
                start = Position{
                }
 
                start = Position{


@@ -122,7 +122,7 @@ func (t *Tree) open(ctx context.Context, total int) error {
                }
        }
 
                }
        }
 

-       return t.journal.Scan(ctx, start, func(line []byte, place Position) error {
+       err = t.journal.Scan(ctx, start, func(line []byte, place Position) error {

                var node Node
 
                err := json.Unmarshal(line, &node)
                var node Node
 
                err := json.Unmarshal(line, &node)


@@ -132,15 +132,25 @@ func (t *Tree) open(ctx context.Context, total int) error {
 
                err = t.check(node)
                if err != nil {
 
                err = t.check(node)
                if err != nil {

-                       return err
+                       return fmt.Errorf("validate scanned node: %w", err)

                }
 
                if nodeEnd(node.Level, node.Index) > total {
                        return fmt.Errorf("node (%d, %d) covers messages after the log", node.Level, node.Index)
                }
 
                }
 
                if nodeEnd(node.Level, node.Index) > total {
                        return fmt.Errorf("node (%d, %d) covers messages after the log", node.Level, node.Index)
                }
 

-               return t.addIndex(NodeKey{node.Level, node.Index}, place)
+               err = t.addIndex(NodeKey{node.Level, node.Index}, place)
+               if err != nil {
+                       return fmt.Errorf("index scanned node: %w", err)
+               }
+
+               return nil

        })
        })

+       if err != nil {
+               return fmt.Errorf("scan node journal: %w", err)
+       }
+
+       return nil

 }
 
 // Close closes the files of the tree.
 }
 
 // Close closes the files of the tree.


@@ -182,13 +192,18 @@ func (t *Tree) removeLevels() error {
 func (t *Tree) writeLevel(node NodeKey, place Position) error {
        level, err := t.level(node.level)
        if err != nil {
 func (t *Tree) writeLevel(node NodeKey, place Position) error {
        level, err := t.level(node.level)
        if err != nil {

-               return err
+               return fmt.Errorf("open node level index: %w", err)

        }
 
        record := make([]byte, positionBytes)
        encodePosition(record, place)
 
        }
 
        record := make([]byte, positionBytes)
        encodePosition(record, place)
 

-       return level.Write(node.index, record)
+       err = level.Write(node.index, record)
+       if err != nil {
+               return fmt.Errorf("write node level record: %w", err)
+       }
+
+       return nil

 }
 
 // check makes sure that node is valid and can join the tree now.
 }
 
 // check makes sure that node is valid and can join the tree now.


@@ -203,7 +218,7 @@ func (t *Tree) check(node Node) error {
 
        built, err := t.Built(node.Level, node.Index)
        if err != nil {
 
        built, err := t.Built(node.Level, node.Index)
        if err != nil {

-               return err
+               return fmt.Errorf("check target node: %w", err)

        }
 
        if built {
        }
 
        if built {


@@ -217,7 +232,7 @@ func (t *Tree) check(node Node) error {
        for _, index := range []int{2 * node.Index, 2*node.Index + 1} {
                built, err = t.Built(node.Level-1, index)
                if err != nil {
        for _, index := range []int{2 * node.Index, 2*node.Index + 1} {
                built, err = t.Built(node.Level-1, index)
                if err != nil {

-                       return err
+                       return fmt.Errorf("check child node: %w", err)

                }
 
                if !built {
                }
 
                if !built {


@@ -237,10 +252,15 @@ func (t *Tree) addIndex(node NodeKey, place Position) error {
 
        err := t.order.Append(record)
        if err != nil {
 
        err := t.order.Append(record)
        if err != nil {

-               return err
+               return fmt.Errorf("append node order: %w", err)
+       }
+
+       err = t.writeLevel(node, place)
+       if err != nil {
+               return fmt.Errorf("write indexed node level: %w", err)

        }
 
        }
 

-       return t.writeLevel(node, place)
+       return nil

 }
 
 // level returns the index of level, and opens it at first use.
 }
 
 // level returns the index of level, and opens it at first use.


@@ -274,12 +294,16 @@ func (t *Tree) place(level, index int) (Position, bool, error) {
 
        records, err := t.level(level)
        if err != nil {
 
        records, err := t.level(level)
        if err != nil {

-               return Position{}, false, err
+               return Position{}, false, fmt.Errorf("open node level index: %w", err)

        }
 
        record, found, err := records.Read(index)
        if err != nil || !found {
        }
 
        record, found, err := records.Read(index)
        if err != nil || !found {

-               return Position{}, false, err
+               if err != nil {
+                       return Position{}, false, fmt.Errorf("read node level record: %w", err)
+               }
+
+               return Position{}, false, nil

        }
 
        place := decodePosition(record)
        }
 
        place := decodePosition(record)


@@ -295,7 +319,7 @@ func (t *Tree) Add(node Node) error {
 
        err := t.check(node)
        if err != nil {
 
        err := t.check(node)
        if err != nil {

-               return err
+               return fmt.Errorf("validate node: %w", err)

        }
 
        line, err := json.Marshal(node)
        }
 
        line, err := json.Marshal(node)


@@ -320,7 +344,11 @@ func (t *Tree) Add(node Node) error {
 func (t *Tree) Node(level, index int) (Node, bool, error) {
        place, built, err := t.place(level, index)
        if err != nil || !built {
 func (t *Tree) Node(level, index int) (Node, bool, error) {
        place, built, err := t.place(level, index)
        if err != nil || !built {

-               return Node{}, false, err
+               if err != nil {
+                       return Node{}, false, fmt.Errorf("find node position: %w", err)
+               }
+
+               return Node{}, false, nil

        }
 
        line, err := t.journal.Read(place)
        }
 
        line, err := t.journal.Read(place)