54dbd297304212367b4e98920f393e87d323509b
[ira/wip.git] / source4 / dsdb / repl / drepl_out_pull.c
1 /* 
2    Unix SMB/CIFS mplementation.
3    DSDB replication service outgoing Pull-Replication
4    
5    Copyright (C) Stefan Metzmacher 2007
6     
7    This program is free software; you can redistribute it and/or modify
8    it under the terms of the GNU General Public License as published by
9    the Free Software Foundation; either version 3 of the License, or
10    (at your option) any later version.
11    
12    This program is distributed in the hope that it will be useful,
13    but WITHOUT ANY WARRANTY; without even the implied warranty of
14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15    GNU General Public License for more details.
16    
17    You should have received a copy of the GNU General Public License
18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
19    
20 */
21
22 #include "includes.h"
23 #include "dsdb/samdb/samdb.h"
24 #include "auth/auth.h"
25 #include "smbd/service.h"
26 #include "lib/events/events.h"
27 #include "lib/messaging/irpc.h"
28 #include "dsdb/repl/drepl_service.h"
29 #include "lib/ldb/include/ldb_errors.h"
30 #include "../lib/util/dlinklist.h"
31 #include "librpc/gen_ndr/ndr_misc.h"
32 #include "librpc/gen_ndr/ndr_drsuapi.h"
33 #include "librpc/gen_ndr/ndr_drsblobs.h"
34 #include "libcli/composite/composite.h"
35
36 static WERROR dreplsrv_schedule_partition_pull_source(struct dreplsrv_service *s,
37                                                       struct dreplsrv_partition *p,
38                                                       struct dreplsrv_partition_source_dsa *source,
39                                                       TALLOC_CTX *mem_ctx)
40 {
41         struct dreplsrv_out_operation *op;
42
43         op = talloc_zero(mem_ctx, struct dreplsrv_out_operation);
44         W_ERROR_HAVE_NO_MEMORY(op);
45
46         op->service     = s;
47         op->source_dsa  = source;
48
49         DLIST_ADD_END(s->ops.pending, op, struct dreplsrv_out_operation *);
50         talloc_steal(s, op);
51         return WERR_OK;
52 }
53
54 static WERROR dreplsrv_schedule_partition_pull(struct dreplsrv_service *s,
55                                                struct dreplsrv_partition *p,
56                                                TALLOC_CTX *mem_ctx)
57 {
58         WERROR status;
59         struct dreplsrv_partition_source_dsa *cur;
60
61         for (cur = p->sources; cur; cur = cur->next) {
62                 status = dreplsrv_schedule_partition_pull_source(s, p, cur, mem_ctx);
63                 W_ERROR_NOT_OK_RETURN(status);
64         }
65
66         return WERR_OK;
67 }
68
69 WERROR dreplsrv_schedule_pull_replication(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx)
70 {
71         WERROR status;
72         struct dreplsrv_partition *p;
73
74         for (p = s->partitions; p; p = p->next) {
75                 status = dreplsrv_schedule_partition_pull(s, p, mem_ctx);
76                 W_ERROR_NOT_OK_RETURN(status);
77         }
78
79         return WERR_OK;
80 }
81
82
83 /* force an immediate of the specified partition by GUID  */
84 WERROR dreplsrv_schedule_partition_pull_by_guid(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx,
85                                                 struct GUID *guid)
86 {
87         struct dreplsrv_partition *p;
88         
89         for (p = s->partitions; p; p = p->next) {
90                 if (GUID_compare(&p->nc.guid, guid) == 0) {
91                         return dreplsrv_schedule_partition_pull(s, p, mem_ctx);
92                 }
93         }
94
95         return WERR_NOT_FOUND;
96 }
97
98 static void dreplsrv_pending_op_callback(struct dreplsrv_out_operation *op)
99 {
100         struct repsFromTo1 *rf = op->source_dsa->repsFrom1;
101         struct dreplsrv_service *s = op->service;
102         time_t t;
103         NTTIME now;
104
105         t = time(NULL);
106         unix_to_nt_time(&now, t);
107
108         rf->result_last_attempt = dreplsrv_op_pull_source_recv(op->creq);
109         if (W_ERROR_IS_OK(rf->result_last_attempt)) {
110                 rf->consecutive_sync_failures   = 0;
111                 rf->last_success                = now;
112                 DEBUG(2,("dreplsrv_op_pull_source(%s)\n",
113                         win_errstr(rf->result_last_attempt)));
114                 goto done;
115         }
116
117         rf->consecutive_sync_failures++;
118
119         DEBUG(1,("dreplsrv_op_pull_source(%s/%s) failures[%u]\n",
120                 win_errstr(rf->result_last_attempt),
121                 nt_errstr(werror_to_ntstatus(rf->result_last_attempt)),
122                 rf->consecutive_sync_failures));
123
124 done:
125         talloc_free(op);
126         s->ops.current = NULL;
127         dreplsrv_run_pending_ops(s);
128 }
129
130 static void dreplsrv_pending_op_callback_creq(struct composite_context *creq)
131 {
132         struct dreplsrv_out_operation *op = talloc_get_type(creq->async.private_data,
133                                                            struct dreplsrv_out_operation);
134         dreplsrv_pending_op_callback(op);
135 }
136
137 void dreplsrv_run_pending_ops(struct dreplsrv_service *s)
138 {
139         struct dreplsrv_out_operation *op;
140         time_t t;
141         NTTIME now;
142
143         if (s->ops.current) {
144                 /* if there's still one running, we're done */
145                 return;
146         }
147
148         if (!s->ops.pending) {
149                 /* if there're no pending operations, we're done */
150                 return;
151         }
152
153         t = time(NULL);
154         unix_to_nt_time(&now, t);
155
156         op = s->ops.pending;
157         s->ops.current = op;
158         DLIST_REMOVE(s->ops.pending, op);
159
160         op->source_dsa->repsFrom1->last_attempt = now;
161
162         op->creq = dreplsrv_op_pull_source_send(op);
163         if (!op->creq) {
164                 dreplsrv_pending_op_callback(op);
165                 return;
166         }
167
168         op->creq->async.fn              = dreplsrv_pending_op_callback_creq;
169         op->creq->async.private_data    = op;
170 }