You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

273 lines
6.6 KiB

  1. package lite
  2. import (
  3. "fmt"
  4. "regexp"
  5. "strconv"
  6. amino "github.com/tendermint/go-amino"
  7. cryptoamino "github.com/tendermint/tendermint/crypto/encoding/amino"
  8. log "github.com/tendermint/tendermint/libs/log"
  9. lerr "github.com/tendermint/tendermint/lite/errors"
  10. "github.com/tendermint/tendermint/types"
  11. dbm "github.com/tendermint/tm-db"
  12. )
  13. var _ PersistentProvider = (*DBProvider)(nil)
  14. // DBProvider stores commits and validator sets in a DB.
  15. type DBProvider struct {
  16. logger log.Logger
  17. label string
  18. db dbm.DB
  19. cdc *amino.Codec
  20. limit int
  21. }
  22. func NewDBProvider(label string, db dbm.DB) *DBProvider {
  23. // NOTE: when debugging, this type of construction might be useful.
  24. //db = dbm.NewDebugDB("db provider "+cmn.RandStr(4), db)
  25. cdc := amino.NewCodec()
  26. cryptoamino.RegisterAmino(cdc)
  27. dbp := &DBProvider{
  28. logger: log.NewNopLogger(),
  29. label: label,
  30. db: db,
  31. cdc: cdc,
  32. }
  33. return dbp
  34. }
  35. func (dbp *DBProvider) SetLogger(logger log.Logger) {
  36. dbp.logger = logger.With("label", dbp.label)
  37. }
  38. func (dbp *DBProvider) SetLimit(limit int) *DBProvider {
  39. dbp.limit = limit
  40. return dbp
  41. }
  42. // Implements PersistentProvider.
  43. func (dbp *DBProvider) SaveFullCommit(fc FullCommit) error {
  44. dbp.logger.Info("DBProvider.SaveFullCommit()...", "fc", fc)
  45. batch := dbp.db.NewBatch()
  46. defer batch.Close()
  47. // Save the fc.validators.
  48. // We might be overwriting what we already have, but
  49. // it makes the logic easier for now.
  50. vsKey := validatorSetKey(fc.ChainID(), fc.Height())
  51. vsBz, err := dbp.cdc.MarshalBinaryLengthPrefixed(fc.Validators)
  52. if err != nil {
  53. return err
  54. }
  55. batch.Set(vsKey, vsBz)
  56. // Save the fc.NextValidators.
  57. nvsKey := validatorSetKey(fc.ChainID(), fc.Height()+1)
  58. nvsBz, err := dbp.cdc.MarshalBinaryLengthPrefixed(fc.NextValidators)
  59. if err != nil {
  60. return err
  61. }
  62. batch.Set(nvsKey, nvsBz)
  63. // Save the fc.SignedHeader
  64. shKey := signedHeaderKey(fc.ChainID(), fc.Height())
  65. shBz, err := dbp.cdc.MarshalBinaryLengthPrefixed(fc.SignedHeader)
  66. if err != nil {
  67. return err
  68. }
  69. batch.Set(shKey, shBz)
  70. // And write sync.
  71. batch.WriteSync()
  72. // Garbage collect.
  73. // TODO: optimize later.
  74. if dbp.limit > 0 {
  75. dbp.deleteAfterN(fc.ChainID(), dbp.limit)
  76. }
  77. return nil
  78. }
  79. // Implements Provider.
  80. func (dbp *DBProvider) LatestFullCommit(chainID string, minHeight, maxHeight int64) (
  81. FullCommit, error) {
  82. dbp.logger.Info("DBProvider.LatestFullCommit()...",
  83. "chainID", chainID, "minHeight", minHeight, "maxHeight", maxHeight)
  84. if minHeight <= 0 {
  85. minHeight = 1
  86. }
  87. if maxHeight == 0 {
  88. maxHeight = 1<<63 - 1
  89. }
  90. itr := dbp.db.ReverseIterator(
  91. signedHeaderKey(chainID, minHeight),
  92. append(signedHeaderKey(chainID, maxHeight), byte(0x00)),
  93. )
  94. defer itr.Close()
  95. for itr.Valid() {
  96. key := itr.Key()
  97. _, _, ok := parseSignedHeaderKey(key)
  98. if !ok {
  99. // Skip over other keys.
  100. itr.Next()
  101. continue
  102. } else {
  103. // Found the latest full commit signed header.
  104. shBz := itr.Value()
  105. sh := types.SignedHeader{}
  106. err := dbp.cdc.UnmarshalBinaryLengthPrefixed(shBz, &sh)
  107. if err != nil {
  108. return FullCommit{}, err
  109. } else {
  110. lfc, err := dbp.fillFullCommit(sh)
  111. if err == nil {
  112. dbp.logger.Info("DBProvider.LatestFullCommit() found latest.", "height", lfc.Height())
  113. return lfc, nil
  114. } else {
  115. dbp.logger.Error("DBProvider.LatestFullCommit() got error", "lfc", lfc)
  116. dbp.logger.Error(fmt.Sprintf("%+v", err))
  117. return lfc, err
  118. }
  119. }
  120. }
  121. }
  122. return FullCommit{}, lerr.ErrCommitNotFound()
  123. }
  124. func (dbp *DBProvider) ValidatorSet(chainID string, height int64) (valset *types.ValidatorSet, err error) {
  125. return dbp.getValidatorSet(chainID, height)
  126. }
  127. func (dbp *DBProvider) getValidatorSet(chainID string, height int64) (valset *types.ValidatorSet, err error) {
  128. vsBz := dbp.db.Get(validatorSetKey(chainID, height))
  129. if vsBz == nil {
  130. err = lerr.ErrUnknownValidators(chainID, height)
  131. return
  132. }
  133. err = dbp.cdc.UnmarshalBinaryLengthPrefixed(vsBz, &valset)
  134. if err != nil {
  135. return
  136. }
  137. // To test deep equality. This makes it easier to test for e.g. valset
  138. // equivalence using assert.Equal (tests for deep equality) in our tests,
  139. // which also tests for unexported/private field equivalence.
  140. valset.TotalVotingPower()
  141. return
  142. }
  143. func (dbp *DBProvider) fillFullCommit(sh types.SignedHeader) (FullCommit, error) {
  144. var chainID = sh.ChainID
  145. var height = sh.Height
  146. var valset, nextValset *types.ValidatorSet
  147. // Load the validator set.
  148. valset, err := dbp.getValidatorSet(chainID, height)
  149. if err != nil {
  150. return FullCommit{}, err
  151. }
  152. // Load the next validator set.
  153. nextValset, err = dbp.getValidatorSet(chainID, height+1)
  154. if err != nil {
  155. return FullCommit{}, err
  156. }
  157. // Return filled FullCommit.
  158. return FullCommit{
  159. SignedHeader: sh,
  160. Validators: valset,
  161. NextValidators: nextValset,
  162. }, nil
  163. }
  164. func (dbp *DBProvider) deleteAfterN(chainID string, after int) error {
  165. dbp.logger.Info("DBProvider.deleteAfterN()...", "chainID", chainID, "after", after)
  166. itr := dbp.db.ReverseIterator(
  167. signedHeaderKey(chainID, 1),
  168. append(signedHeaderKey(chainID, 1<<63-1), byte(0x00)),
  169. )
  170. defer itr.Close()
  171. var lastHeight int64 = 1<<63 - 1
  172. var numSeen = 0
  173. var numDeleted = 0
  174. for itr.Valid() {
  175. key := itr.Key()
  176. _, height, ok := parseChainKeyPrefix(key)
  177. if !ok {
  178. return fmt.Errorf("unexpected key %v", key)
  179. } else {
  180. if height < lastHeight {
  181. lastHeight = height
  182. numSeen += 1
  183. }
  184. if numSeen > after {
  185. dbp.db.Delete(key)
  186. numDeleted += 1
  187. }
  188. }
  189. itr.Next()
  190. }
  191. dbp.logger.Info(fmt.Sprintf("DBProvider.deleteAfterN() deleted %v items", numDeleted))
  192. return nil
  193. }
  194. //----------------------------------------
  195. // key encoding
  196. func signedHeaderKey(chainID string, height int64) []byte {
  197. return []byte(fmt.Sprintf("%s/%010d/sh", chainID, height))
  198. }
  199. func validatorSetKey(chainID string, height int64) []byte {
  200. return []byte(fmt.Sprintf("%s/%010d/vs", chainID, height))
  201. }
  202. //----------------------------------------
  203. // key parsing
  204. var keyPattern = regexp.MustCompile(`^([^/]+)/([0-9]*)/(.*)$`)
  205. func parseKey(key []byte) (chainID string, height int64, part string, ok bool) {
  206. submatch := keyPattern.FindSubmatch(key)
  207. if submatch == nil {
  208. return "", 0, "", false
  209. }
  210. chainID = string(submatch[1])
  211. heightStr := string(submatch[2])
  212. heightInt, err := strconv.Atoi(heightStr)
  213. if err != nil {
  214. return "", 0, "", false
  215. }
  216. height = int64(heightInt)
  217. part = string(submatch[3])
  218. ok = true // good!
  219. return
  220. }
  221. func parseSignedHeaderKey(key []byte) (chainID string, height int64, ok bool) {
  222. var part string
  223. chainID, height, part, ok = parseKey(key)
  224. if part != "sh" {
  225. return "", 0, false
  226. }
  227. return
  228. }
  229. func parseChainKeyPrefix(key []byte) (chainID string, height int64, ok bool) {
  230. chainID, height, _, ok = parseKey(key)
  231. return
  232. }