forked from ipfs-cluster/ipfs-cluster
-
Notifications
You must be signed in to change notification settings - Fork 0
/
log_op.go
77 lines (67 loc) · 1.71 KB
/
log_op.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
package raft
import (
"errors"
"github.com/ipfs/ipfs-cluster/api"
"github.com/ipfs/ipfs-cluster/state"
consensus "github.com/libp2p/go-libp2p-consensus"
)
// Type of consensus operation
const (
LogOpPin = iota + 1
LogOpUnpin
)
// LogOpType expresses the type of a consensus Operation
type LogOpType int
// LogOp represents an operation for the OpLogConsensus system.
// It implements the consensus.Op interface and it is used by the
// Consensus component.
type LogOp struct {
Cid api.PinSerial
Type LogOpType
consensus *Consensus
}
// ApplyTo applies the operation to the State
func (op *LogOp) ApplyTo(cstate consensus.State) (consensus.State, error) {
state, ok := cstate.(state.State)
var err error
if !ok {
// Should never be here
panic("received unexpected state type")
}
switch op.Type {
case LogOpPin:
err = state.Add(op.Cid.ToPin())
if err != nil {
goto ROLLBACK
}
// Async, we let the PinTracker take care of any problems
op.consensus.rpcClient.Go("",
"Cluster",
"Track",
op.Cid,
&struct{}{},
nil)
case LogOpUnpin:
err = state.Rm(op.Cid.ToPin().Cid)
if err != nil {
goto ROLLBACK
}
// Async, we let the PinTracker take care of any problems
op.consensus.rpcClient.Go("",
"Cluster",
"Untrack",
op.Cid,
&struct{}{},
nil)
default:
logger.Error("unknown LogOp type. Ignoring")
}
return state, nil
ROLLBACK:
// We failed to apply the operation to the state
// and therefore we need to request a rollback to the
// cluster to the previous state. This operation can only be performed
// by the cluster leader.
logger.Error("Rollbacks are not implemented")
return nil, errors.New("a rollback may be necessary. Reason: " + err.Error())
}