The worker job runner

The worker's StartJob implementation is slightly more complicated as we need to load the vertices and edges that correspond to the UUID range assigned to us by the master node. Fortunately, we have already written all of the required bits of code in Chapter 11, Splitting Monoliths into Microservices, so we can just go ahead and invoke the loading functions with the appropriate arguments:

func (n *WorkerNode) StartJob(jobDetails job.Details, execFactory bspgraph.ExecutorFactory) (*bspgraph.Executor, error) {
    n.jobStartedAt = time.Now()
    if err := n.calculator.Graph().Reset(); err != nil {
        return nil, err
    } else if err := n.loadLinks(jobDetails.PartitionFromID, jobDetails.PartitionToID, jobDetails.CreatedAt); err != nil {
        return nil, err
    } else if err := n.loadEdges(jobDetails.PartitionFromID, jobDetails.PartitionToID, jobDetails.CreatedAt); err != nil {
        return nil, err
    }
    n.graphPopulateTime = time.Since(n.jobStartedAt)

    n.scoreCalculationStartedAt = time.Now()
    n.calculator.SetExecutorFactory(execFactory)
    return n.calculator.Executor(), nil
}

The CompleteJob method contains the necessary logic for updating the Links 'R' Us document index with the fresh PageRank scores that we just calculated. Let's take a look at its implementation:

func (n *WorkerNode) CompleteJob(_ job.Details) error {
    scoreCalculationTime := time.Since(n.scoreCalculationStartedAt)
    tick := time.Now()
    if err := n.calculator.Scores(n.persistScore); err != nil {
        return err
    }
    scorePersistTime := time.Since(tick)
    n.cfg.Logger.WithFields(logrus.Fields{
        "local_link_count":       len(n.calculator.Graph().Vertices()),
        "total_link_count":       n.calculator.Graph().Aggregator("page_count").Get(),
        "graph_populate_time":    n.graphPopulateTime.String(),
        "score_calculation_time": scoreCalculationTime.String(),
        "score_persist_time":     scorePersistTime.String(),
        "total_pass_time":        time.Since(n.jobStartedAt).String(),
    }).Info("completed PageRank update pass")
    return nil
}

The preceding block of code for persisting the calculation results should seem familiar to you as it has been copied verbatim from Chapter 11, Splitting Monoliths into Microservices. The Scores convenience method iterates the graph vertices and invokes the persistScore callback with the vertex ID and PageRank score as arguments.

The persistScore callback (shown as follows) is a simple wrapper for mapping the vertex ID into a UUID value and calling the UpdateScore method of the Links 'R' Us document index component:

func (n *WorkerNode) persistScore(vertexID string, score float64) error {
    linkID, err := uuid.Parse(vertexID)
    if err != nil {
        return err
    }
    return n.cfg.IndexAPI.UpdateScore(linkID, score)
}

Similar to the master job runner implementation, the worker's AbortJob method is also an empty stub. To keep our implementation as lean as possible, we won't bother rolling back any already persisted score changes if any of the other workers fails after the local worker has already completed the job. Since the PageRank scores are periodically re-calculated, we expect them to be eventually consistent.