wrepl_server: make 'use inform' the default and autofallback against old servers
[kai/samba.git] / source4 / wrepl_server / wrepl_out_helpers.c
1 /* 
2    Unix SMB/CIFS implementation.
3    
4    WINS Replication server
5    
6    Copyright (C) Stefan Metzmacher      2005
7    
8    This program is free software; you can redistribute it and/or modify
9    it under the terms of the GNU General Public License as published by
10    the Free Software Foundation; either version 3 of the License, or
11    (at your option) any later version.
12    
13    This program is distributed in the hope that it will be useful,
14    but WITHOUT ANY WARRANTY; without even the implied warranty of
15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16    GNU General Public License for more details.
17    
18    You should have received a copy of the GNU General Public License
19    along with this program.  If not, see <http://www.gnu.org/licenses/>.
20 */
21
22 #include "includes.h"
23 #include "lib/events/events.h"
24 #include "lib/socket/socket.h"
25 #include "smbd/service_task.h"
26 #include "smbd/service_stream.h"
27 #include "librpc/gen_ndr/winsrepl.h"
28 #include "wrepl_server/wrepl_server.h"
29 #include "nbt_server/wins/winsdb.h"
30 #include "libcli/composite/composite.h"
31 #include "libcli/wrepl/winsrepl.h"
32 #include "libcli/resolve/resolve.h"
33 #include "param/param.h"
34
35 enum wreplsrv_out_connect_stage {
36         WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET,
37         WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX,
38         WREPLSRV_OUT_CONNECT_STAGE_DONE
39 };
40
41 struct wreplsrv_out_connect_state {
42         enum wreplsrv_out_connect_stage stage;
43         struct composite_context *c;
44         struct wrepl_request *req;
45         struct composite_context *c_req;
46         struct wrepl_associate assoc_io;
47         enum winsrepl_partner_type type;
48         struct wreplsrv_out_connection *wreplconn;
49 };
50
51 static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
52 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
53
54 static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
55 {
56         NTSTATUS status;
57
58         status = wrepl_connect_recv(state->c_req);
59         NT_STATUS_NOT_OK_RETURN(status);
60
61         state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
62         NT_STATUS_HAVE_NO_MEMORY(state->req);
63
64         state->req->async.fn            = wreplsrv_out_connect_handler_req;
65         state->req->async.private       = state;
66
67         state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
68
69         return NT_STATUS_OK;
70 }
71
72 static NTSTATUS wreplsrv_out_connect_wait_assoc_ctx(struct wreplsrv_out_connect_state *state)
73 {
74         NTSTATUS status;
75
76         status = wrepl_associate_recv(state->req, &state->assoc_io);
77         NT_STATUS_NOT_OK_RETURN(status);
78
79         state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
80         state->wreplconn->assoc_ctx.peer_major = state->assoc_io.out.major_version;
81
82         if (state->type == WINSREPL_PARTNER_PUSH) {
83                 if (state->wreplconn->assoc_ctx.peer_major >= 5) {
84                         state->wreplconn->partner->push.wreplconn = state->wreplconn;
85                         talloc_steal(state->wreplconn->partner, state->wreplconn);
86                 } else {
87                         state->type = WINSREPL_PARTNER_NONE;
88                 }
89         } else if (state->type == WINSREPL_PARTNER_PULL) {
90                 state->wreplconn->partner->pull.wreplconn = state->wreplconn;
91                 talloc_steal(state->wreplconn->partner, state->wreplconn);
92         }
93
94         state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
95
96         return NT_STATUS_OK;
97 }
98
99 static void wreplsrv_out_connect_handler(struct wreplsrv_out_connect_state *state)
100 {
101         struct composite_context *c = state->c;
102
103         switch (state->stage) {
104         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET:
105                 c->status = wreplsrv_out_connect_wait_socket(state);
106                 break;
107         case WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX:
108                 c->status = wreplsrv_out_connect_wait_assoc_ctx(state);
109                 c->state  = COMPOSITE_STATE_DONE;
110                 break;
111         case WREPLSRV_OUT_CONNECT_STAGE_DONE:
112                 c->status = NT_STATUS_INTERNAL_ERROR;
113         }
114
115         if (!NT_STATUS_IS_OK(c->status)) {
116                 c->state = COMPOSITE_STATE_ERROR;
117         }
118
119         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
120                 c->async.fn(c);
121         }
122 }
123
124 static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
125 {
126         struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
127                                                    struct wreplsrv_out_connect_state);
128         wreplsrv_out_connect_handler(state);
129         return;
130 }
131
132 static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
133 {
134         struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private,
135                                                    struct wreplsrv_out_connect_state);
136         wreplsrv_out_connect_handler(state);
137         return;
138 }
139
140 static struct composite_context *wreplsrv_out_connect_send(struct wreplsrv_partner *partner,
141                                                            enum winsrepl_partner_type type,
142                                                            struct wreplsrv_out_connection *wreplconn)
143 {
144         struct composite_context *c = NULL;
145         struct wreplsrv_service *service = partner->service;
146         struct wreplsrv_out_connect_state *state = NULL;
147         struct wreplsrv_out_connection **wreplconnp = &wreplconn;
148         bool cached_connection = false;
149
150         c = talloc_zero(partner, struct composite_context);
151         if (!c) goto failed;
152
153         state = talloc_zero(c, struct wreplsrv_out_connect_state);
154         if (!state) goto failed;
155         state->c        = c;
156         state->type     = type;
157
158         c->state        = COMPOSITE_STATE_IN_PROGRESS;
159         c->event_ctx    = service->task->event_ctx;
160         c->private_data = state;
161
162         if (type == WINSREPL_PARTNER_PUSH) {
163                 cached_connection       = true;
164                 wreplconn               = partner->push.wreplconn;
165                 wreplconnp              = &partner->push.wreplconn;
166         } else if (type == WINSREPL_PARTNER_PULL) {
167                 cached_connection       = true;
168                 wreplconn               = partner->pull.wreplconn;
169                 wreplconnp              = &partner->pull.wreplconn;
170         }
171
172         /* we have a connection already, so use it */
173         if (wreplconn) {
174                 if (!wreplconn->sock->dead) {
175                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
176                         state->wreplconn= wreplconn;
177                         composite_done(c);
178                         return c;
179                 } else if (!cached_connection) {
180                         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_DONE;
181                         state->wreplconn= NULL;
182                         composite_done(c);
183                         return c;
184                 } else {
185                         talloc_free(wreplconn);
186                         *wreplconnp = NULL;
187                 }
188         }
189
190         wreplconn = talloc_zero(state, struct wreplsrv_out_connection);
191         if (!wreplconn) goto failed;
192
193         wreplconn->service      = service;
194         wreplconn->partner      = partner;
195         wreplconn->sock         = wrepl_socket_init(wreplconn, service->task->event_ctx, lp_iconv_convenience(service->task->lp_ctx));
196         if (!wreplconn->sock) goto failed;
197
198         state->stage    = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
199         state->wreplconn= wreplconn;
200         state->c_req    = wrepl_connect_send(wreplconn->sock,
201                                              partner->our_address?partner->our_address:wrepl_best_ip(service->task->lp_ctx, partner->address),
202                                              partner->address);
203         if (!state->c_req) goto failed;
204
205         state->c_req->async.fn                  = wreplsrv_out_connect_handler_creq;
206         state->c_req->async.private_data        = state;
207
208         return c;
209 failed:
210         talloc_free(c);
211         return NULL;
212 }
213
214 static NTSTATUS wreplsrv_out_connect_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
215                                           struct wreplsrv_out_connection **wreplconn)
216 {
217         NTSTATUS status;
218
219         status = composite_wait(c);
220
221         if (NT_STATUS_IS_OK(status)) {
222                 struct wreplsrv_out_connect_state *state = talloc_get_type(c->private_data,
223                                                            struct wreplsrv_out_connect_state);
224                 if (state->wreplconn) {
225                         *wreplconn = talloc_reference(mem_ctx, state->wreplconn);
226                         if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
227                 } else {
228                         status = NT_STATUS_INVALID_CONNECTION;
229                 }
230         }
231
232         talloc_free(c);
233         return status;
234         
235 }
236
237 struct wreplsrv_pull_table_io {
238         struct {
239                 struct wreplsrv_partner *partner;
240                 uint32_t num_owners;
241                 struct wrepl_wins_owner *owners;
242         } in;
243         struct {
244                 uint32_t num_owners;
245                 struct wrepl_wins_owner *owners;
246         } out;
247 };
248
249 enum wreplsrv_pull_table_stage {
250         WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION,
251         WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY,
252         WREPLSRV_PULL_TABLE_STAGE_DONE
253 };
254
255 struct wreplsrv_pull_table_state {
256         enum wreplsrv_pull_table_stage stage;
257         struct composite_context *c;
258         struct wrepl_request *req;
259         struct wrepl_pull_table table_io;
260         struct wreplsrv_pull_table_io *io;
261         struct composite_context *creq;
262         struct wreplsrv_out_connection *wreplconn;
263 };
264
265 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
266
267 static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
268 {
269         NTSTATUS status;
270
271         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
272         NT_STATUS_NOT_OK_RETURN(status);
273
274         state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
275         state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
276         NT_STATUS_HAVE_NO_MEMORY(state->req);
277
278         state->req->async.fn            = wreplsrv_pull_table_handler_req;
279         state->req->async.private       = state;
280
281         state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
282
283         return NT_STATUS_OK;
284 }
285
286 static NTSTATUS wreplsrv_pull_table_wait_table_reply(struct wreplsrv_pull_table_state *state)
287 {
288         NTSTATUS status;
289
290         status = wrepl_pull_table_recv(state->req, state, &state->table_io);
291         NT_STATUS_NOT_OK_RETURN(status);
292
293         state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
294
295         return NT_STATUS_OK;
296 }
297
298 static void wreplsrv_pull_table_handler(struct wreplsrv_pull_table_state *state)
299 {
300         struct composite_context *c = state->c;
301
302         switch (state->stage) {
303         case WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION:
304                 c->status = wreplsrv_pull_table_wait_connection(state);
305                 break;
306         case WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY:
307                 c->status = wreplsrv_pull_table_wait_table_reply(state);
308                 c->state  = COMPOSITE_STATE_DONE;
309                 break;
310         case WREPLSRV_PULL_TABLE_STAGE_DONE:
311                 c->status = NT_STATUS_INTERNAL_ERROR;
312         }
313
314         if (!NT_STATUS_IS_OK(c->status)) {
315                 c->state = COMPOSITE_STATE_ERROR;
316         }
317
318         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
319                 c->async.fn(c);
320         }
321 }
322
323 static void wreplsrv_pull_table_handler_creq(struct composite_context *creq)
324 {
325         struct wreplsrv_pull_table_state *state = talloc_get_type(creq->async.private_data,
326                                                   struct wreplsrv_pull_table_state);
327         wreplsrv_pull_table_handler(state);
328         return;
329 }
330
331 static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
332 {
333         struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private,
334                                                   struct wreplsrv_pull_table_state);
335         wreplsrv_pull_table_handler(state);
336         return;
337 }
338
339 static struct composite_context *wreplsrv_pull_table_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_table_io *io)
340 {
341         struct composite_context *c = NULL;
342         struct wreplsrv_service *service = io->in.partner->service;
343         struct wreplsrv_pull_table_state *state = NULL;
344
345         c = talloc_zero(mem_ctx, struct composite_context);
346         if (!c) goto failed;
347
348         state = talloc_zero(c, struct wreplsrv_pull_table_state);
349         if (!state) goto failed;
350         state->c        = c;
351         state->io       = io;
352
353         c->state        = COMPOSITE_STATE_IN_PROGRESS;
354         c->event_ctx    = service->task->event_ctx;
355         c->private_data = state;
356
357         if (io->in.num_owners) {
358                 state->table_io.out.num_partners        = io->in.num_owners;
359                 state->table_io.out.partners            = io->in.owners;
360                 state->stage                            = WREPLSRV_PULL_TABLE_STAGE_DONE;
361                 composite_done(c);
362                 return c;
363         }
364
365         state->stage    = WREPLSRV_PULL_TABLE_STAGE_WAIT_CONNECTION;
366         state->creq     = wreplsrv_out_connect_send(io->in.partner, WINSREPL_PARTNER_PULL, NULL);
367         if (!state->creq) goto failed;
368
369         state->creq->async.fn           = wreplsrv_pull_table_handler_creq;
370         state->creq->async.private_data = state;
371
372         return c;
373 failed:
374         talloc_free(c);
375         return NULL;
376 }
377
378 static NTSTATUS wreplsrv_pull_table_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
379                                          struct wreplsrv_pull_table_io *io)
380 {
381         NTSTATUS status;
382
383         status = composite_wait(c);
384
385         if (NT_STATUS_IS_OK(status)) {
386                 struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
387                                                           struct wreplsrv_pull_table_state);
388                 io->out.num_owners      = state->table_io.out.num_partners;
389                 io->out.owners          = talloc_reference(mem_ctx, state->table_io.out.partners);
390         }
391
392         talloc_free(c);
393         return status;  
394 }
395
396 struct wreplsrv_pull_names_io {
397         struct {
398                 struct wreplsrv_partner *partner;
399                 struct wreplsrv_out_connection *wreplconn;
400                 struct wrepl_wins_owner owner;
401         } in;
402         struct {
403                 uint32_t num_names;
404                 struct wrepl_name *names;
405         } out;
406 };
407
408 enum wreplsrv_pull_names_stage {
409         WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION,
410         WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY,
411         WREPLSRV_PULL_NAMES_STAGE_DONE
412 };
413
414 struct wreplsrv_pull_names_state {
415         enum wreplsrv_pull_names_stage stage;
416         struct composite_context *c;
417         struct wrepl_request *req;
418         struct wrepl_pull_names pull_io;
419         struct wreplsrv_pull_names_io *io;
420         struct composite_context *creq;
421         struct wreplsrv_out_connection *wreplconn;
422 };
423
424 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req);
425
426 static NTSTATUS wreplsrv_pull_names_wait_connection(struct wreplsrv_pull_names_state *state)
427 {
428         NTSTATUS status;
429
430         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
431         NT_STATUS_NOT_OK_RETURN(status);
432
433         state->pull_io.in.assoc_ctx     = state->wreplconn->assoc_ctx.peer_ctx;
434         state->pull_io.in.partner       = state->io->in.owner;
435         state->req = wrepl_pull_names_send(state->wreplconn->sock, &state->pull_io);
436         NT_STATUS_HAVE_NO_MEMORY(state->req);
437
438         state->req->async.fn            = wreplsrv_pull_names_handler_req;
439         state->req->async.private       = state;
440
441         state->stage = WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY;
442
443         return NT_STATUS_OK;
444 }
445
446 static NTSTATUS wreplsrv_pull_names_wait_send_reply(struct wreplsrv_pull_names_state *state)
447 {
448         NTSTATUS status;
449
450         status = wrepl_pull_names_recv(state->req, state, &state->pull_io);
451         NT_STATUS_NOT_OK_RETURN(status);
452
453         state->stage = WREPLSRV_PULL_NAMES_STAGE_DONE;
454
455         return NT_STATUS_OK;
456 }
457
458 static void wreplsrv_pull_names_handler(struct wreplsrv_pull_names_state *state)
459 {
460         struct composite_context *c = state->c;
461
462         switch (state->stage) {
463         case WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION:
464                 c->status = wreplsrv_pull_names_wait_connection(state);
465                 break;
466         case WREPLSRV_PULL_NAMES_STAGE_WAIT_SEND_REPLY:
467                 c->status = wreplsrv_pull_names_wait_send_reply(state);
468                 c->state  = COMPOSITE_STATE_DONE;
469                 break;
470         case WREPLSRV_PULL_NAMES_STAGE_DONE:
471                 c->status = NT_STATUS_INTERNAL_ERROR;
472         }
473
474         if (!NT_STATUS_IS_OK(c->status)) {
475                 c->state = COMPOSITE_STATE_ERROR;
476         }
477
478         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
479                 c->async.fn(c);
480         }
481 }
482
483 static void wreplsrv_pull_names_handler_creq(struct composite_context *creq)
484 {
485         struct wreplsrv_pull_names_state *state = talloc_get_type(creq->async.private_data,
486                                                   struct wreplsrv_pull_names_state);
487         wreplsrv_pull_names_handler(state);
488         return;
489 }
490
491 static void wreplsrv_pull_names_handler_req(struct wrepl_request *req)
492 {
493         struct wreplsrv_pull_names_state *state = talloc_get_type(req->async.private,
494                                                   struct wreplsrv_pull_names_state);
495         wreplsrv_pull_names_handler(state);
496         return;
497 }
498
499 static struct composite_context *wreplsrv_pull_names_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_names_io *io)
500 {
501         struct composite_context *c = NULL;
502         struct wreplsrv_service *service = io->in.partner->service;
503         struct wreplsrv_pull_names_state *state = NULL;
504         enum winsrepl_partner_type partner_type = WINSREPL_PARTNER_PULL;
505
506         if (io->in.wreplconn) partner_type = WINSREPL_PARTNER_NONE;
507
508         c = talloc_zero(mem_ctx, struct composite_context);
509         if (!c) goto failed;
510
511         state = talloc_zero(c, struct wreplsrv_pull_names_state);
512         if (!state) goto failed;
513         state->c        = c;
514         state->io       = io;
515
516         c->state        = COMPOSITE_STATE_IN_PROGRESS;
517         c->event_ctx    = service->task->event_ctx;
518         c->private_data = state;
519
520         state->stage    = WREPLSRV_PULL_NAMES_STAGE_WAIT_CONNECTION;
521         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, io->in.wreplconn);
522         if (!state->creq) goto failed;
523
524         state->creq->async.fn           = wreplsrv_pull_names_handler_creq;
525         state->creq->async.private_data = state;
526
527         return c;
528 failed:
529         talloc_free(c);
530         return NULL;
531 }
532
533 static NTSTATUS wreplsrv_pull_names_recv(struct composite_context *c, TALLOC_CTX *mem_ctx,
534                                          struct wreplsrv_pull_names_io *io)
535 {
536         NTSTATUS status;
537
538         status = composite_wait(c);
539
540         if (NT_STATUS_IS_OK(status)) {
541                 struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
542                                                           struct wreplsrv_pull_names_state);
543                 io->out.num_names       = state->pull_io.out.num_names;
544                 io->out.names           = talloc_reference(mem_ctx, state->pull_io.out.names);
545         }
546
547         talloc_free(c);
548         return status;
549         
550 }
551
552 enum wreplsrv_pull_cycle_stage {
553         WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY,
554         WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES,
555         WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC,
556         WREPLSRV_PULL_CYCLE_STAGE_DONE
557 };
558
559 struct wreplsrv_pull_cycle_state {
560         enum wreplsrv_pull_cycle_stage stage;
561         struct composite_context *c;
562         struct wreplsrv_pull_cycle_io *io;
563         struct wreplsrv_pull_table_io table_io;
564         uint32_t current;
565         struct wreplsrv_pull_names_io names_io;
566         struct composite_context *creq;
567         struct wrepl_associate_stop assoc_stop_io;
568         struct wrepl_request *req;
569 };
570
571 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
572 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
573
574 static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
575 {
576         struct wreplsrv_owner *current_owner=NULL;
577         struct wreplsrv_owner *local_owner;
578         uint32_t i;
579         uint64_t old_max_version = 0;
580         bool do_pull = false;
581
582         for (i=state->current; i < state->table_io.out.num_owners; i++) {
583                 current_owner = wreplsrv_find_owner(state->io->in.partner->service,
584                                                     state->io->in.partner->pull.table,
585                                                     state->table_io.out.owners[i].address);
586
587                 local_owner = wreplsrv_find_owner(state->io->in.partner->service,
588                                                   state->io->in.partner->service->table,
589                                                   state->table_io.out.owners[i].address);
590                 /*
591                  * this means we are ourself the current owner,
592                  * and we don't want replicate ourself
593                  */
594                 if (!current_owner) continue;
595
596                 /*
597                  * this means we don't have any records of this owner
598                  * so fetch them
599                  */
600                 if (!local_owner) {
601                         do_pull         = true;
602                         
603                         break;
604                 }
605
606                 /*
607                  * this means the remote partner has some new records of this owner
608                  * fetch them
609                  */
610                 if (current_owner->owner.max_version > local_owner->owner.max_version) {
611                         do_pull         = true;
612                         old_max_version = local_owner->owner.max_version;
613                         break;
614                 }
615         }
616         state->current = i;
617
618         if (do_pull) {
619                 state->names_io.in.partner              = state->io->in.partner;
620                 state->names_io.in.wreplconn            = state->io->in.wreplconn;
621                 state->names_io.in.owner                = current_owner->owner;
622                 state->names_io.in.owner.min_version    = old_max_version + 1;
623                 state->creq = wreplsrv_pull_names_send(state, &state->names_io);
624                 NT_STATUS_HAVE_NO_MEMORY(state->creq);
625
626                 state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
627                 state->creq->async.private_data = state;
628
629                 return STATUS_MORE_ENTRIES;
630         }
631
632         return NT_STATUS_OK;
633 }
634
635 static NTSTATUS wreplsrv_pull_cycle_next_owner_wrapper(struct wreplsrv_pull_cycle_state *state)
636 {
637         NTSTATUS status;
638
639         status = wreplsrv_pull_cycle_next_owner_do_work(state);
640         if (NT_STATUS_IS_OK(status)) {
641                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
642         } else if (NT_STATUS_EQUAL(STATUS_MORE_ENTRIES, status)) {
643                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES;
644                 status = NT_STATUS_OK;
645         }
646
647         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
648                 state->assoc_stop_io.in.assoc_ctx       = state->io->in.wreplconn->assoc_ctx.peer_ctx;
649                 state->assoc_stop_io.in.reason          = 0;
650                 state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
651                 NT_STATUS_HAVE_NO_MEMORY(state->req);
652
653                 state->req->async.fn            = wreplsrv_pull_cycle_handler_req;
654                 state->req->async.private       = state;
655
656                 state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
657         }
658
659         return status;
660 }
661
662 static NTSTATUS wreplsrv_pull_cycle_wait_table_reply(struct wreplsrv_pull_cycle_state *state)
663 {
664         NTSTATUS status;
665         uint32_t i;
666
667         status = wreplsrv_pull_table_recv(state->creq, state, &state->table_io);
668         NT_STATUS_NOT_OK_RETURN(status);
669
670         /* update partner table */
671         for (i=0; i < state->table_io.out.num_owners; i++) {
672                 status = wreplsrv_add_table(state->io->in.partner->service,
673                                             state->io->in.partner, 
674                                             &state->io->in.partner->pull.table,
675                                             state->table_io.out.owners[i].address,
676                                             state->table_io.out.owners[i].max_version);
677                 NT_STATUS_NOT_OK_RETURN(status);
678         }
679
680         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
681         NT_STATUS_NOT_OK_RETURN(status);
682
683         return status;
684 }
685
686 static NTSTATUS wreplsrv_pull_cycle_apply_records(struct wreplsrv_pull_cycle_state *state)
687 {
688         NTSTATUS status;
689
690         status = wreplsrv_apply_records(state->io->in.partner,
691                                         &state->names_io.in.owner,
692                                         state->names_io.out.num_names,
693                                         state->names_io.out.names);
694         NT_STATUS_NOT_OK_RETURN(status);
695
696         talloc_free(state->names_io.out.names);
697         ZERO_STRUCT(state->names_io);
698
699         return NT_STATUS_OK;
700 }
701
702 static NTSTATUS wreplsrv_pull_cycle_wait_send_replies(struct wreplsrv_pull_cycle_state *state)
703 {
704         NTSTATUS status;
705
706         status = wreplsrv_pull_names_recv(state->creq, state, &state->names_io);
707         NT_STATUS_NOT_OK_RETURN(status);
708
709         /*
710          * TODO: this should maybe an async call,
711          *       because we may need some network access
712          *       for conflict resolving
713          */
714         status = wreplsrv_pull_cycle_apply_records(state);
715         NT_STATUS_NOT_OK_RETURN(status);
716
717         status = wreplsrv_pull_cycle_next_owner_wrapper(state);
718         NT_STATUS_NOT_OK_RETURN(status);
719
720         return status;
721 }
722
723 static NTSTATUS wreplsrv_pull_cycle_wait_stop_assoc(struct wreplsrv_pull_cycle_state *state)
724 {
725         NTSTATUS status;
726
727         status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
728         NT_STATUS_NOT_OK_RETURN(status);
729
730         state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
731
732         return status;
733 }
734
735 static void wreplsrv_pull_cycle_handler(struct wreplsrv_pull_cycle_state *state)
736 {
737         struct composite_context *c = state->c;
738
739         switch (state->stage) {
740         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY:
741                 c->status = wreplsrv_pull_cycle_wait_table_reply(state);
742                 break;
743         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_SEND_REPLIES:
744                 c->status = wreplsrv_pull_cycle_wait_send_replies(state);
745                 break;
746         case WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC:
747                 c->status = wreplsrv_pull_cycle_wait_stop_assoc(state);
748                 break;
749         case WREPLSRV_PULL_CYCLE_STAGE_DONE:
750                 c->status = NT_STATUS_INTERNAL_ERROR;
751         }
752
753         if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE) {
754                 c->state  = COMPOSITE_STATE_DONE;
755         }
756
757         if (!NT_STATUS_IS_OK(c->status)) {
758                 c->state = COMPOSITE_STATE_ERROR;
759         }
760
761         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
762                 c->async.fn(c);
763         }
764 }
765
766 static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq)
767 {
768         struct wreplsrv_pull_cycle_state *state = talloc_get_type(creq->async.private_data,
769                                                   struct wreplsrv_pull_cycle_state);
770         wreplsrv_pull_cycle_handler(state);
771         return;
772 }
773
774 static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
775 {
776         struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private,
777                                                   struct wreplsrv_pull_cycle_state);
778         wreplsrv_pull_cycle_handler(state);
779         return;
780 }
781
782 struct composite_context *wreplsrv_pull_cycle_send(TALLOC_CTX *mem_ctx, struct wreplsrv_pull_cycle_io *io)
783 {
784         struct composite_context *c = NULL;
785         struct wreplsrv_service *service = io->in.partner->service;
786         struct wreplsrv_pull_cycle_state *state = NULL;
787
788         c = talloc_zero(mem_ctx, struct composite_context);
789         if (!c) goto failed;
790
791         state = talloc_zero(c, struct wreplsrv_pull_cycle_state);
792         if (!state) goto failed;
793         state->c        = c;
794         state->io       = io;
795
796         c->state        = COMPOSITE_STATE_IN_PROGRESS;
797         c->event_ctx    = service->task->event_ctx;
798         c->private_data = state;
799
800         state->stage    = WREPLSRV_PULL_CYCLE_STAGE_WAIT_TABLE_REPLY;
801         state->table_io.in.partner      = io->in.partner;
802         state->table_io.in.num_owners   = io->in.num_owners;
803         state->table_io.in.owners       = io->in.owners;
804         state->creq = wreplsrv_pull_table_send(state, &state->table_io);
805         if (!state->creq) goto failed;
806
807         state->creq->async.fn           = wreplsrv_pull_cycle_handler_creq;
808         state->creq->async.private_data = state;
809
810         return c;
811 failed:
812         talloc_free(c);
813         return NULL;
814 }
815
816 NTSTATUS wreplsrv_pull_cycle_recv(struct composite_context *c)
817 {
818         NTSTATUS status;
819
820         status = composite_wait(c);
821
822         talloc_free(c);
823         return status;
824 }
825
826 enum wreplsrv_push_notify_stage {
827         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
828         WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM,
829         WREPLSRV_PUSH_NOTIFY_STAGE_DONE
830 };
831
832 struct wreplsrv_push_notify_state {
833         enum wreplsrv_push_notify_stage stage;
834         struct composite_context *c;
835         struct wreplsrv_push_notify_io *io;
836         enum wrepl_replication_cmd command;
837         bool full_table;
838         struct wrepl_send_ctrl ctrl;
839         struct wrepl_request *req;
840         struct wrepl_packet req_packet;
841         struct wrepl_packet *rep_packet;
842         struct composite_context *creq;
843         struct wreplsrv_out_connection *wreplconn;
844 };
845
846 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
847 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
848
849 static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
850 {
851         struct wreplsrv_service *service = state->io->in.partner->service;
852         struct wrepl_packet *req = &state->req_packet;
853         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
854         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
855         struct wreplsrv_in_connection *wrepl_in;
856         NTSTATUS status;
857         struct socket_context *sock;
858         struct packet_context *packet;
859         uint16_t fde_flags;
860
861         /* prepare the outgoing request */
862         req->opcode     = WREPL_OPCODE_BITS;
863         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
864         req->mess_type  = WREPL_REPLICATION;
865
866         repl_out->command = state->command;
867
868         status = wreplsrv_fill_wrepl_table(service, state, table_out,
869                                            service->wins_db->local_owner, state->full_table);
870         NT_STATUS_NOT_OK_RETURN(status);
871
872         /* queue the request */
873         state->req = wrepl_request_send(state->wreplconn->sock, req, NULL);
874         NT_STATUS_HAVE_NO_MEMORY(state->req);
875
876         /*
877          * now we need to convert the wrepl_socket (client connection)
878          * into a wreplsrv_in_connection (server connection), because
879          * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
880          * message is received by the peer.
881          */
882
883         /* steal the socket_context */
884         sock = state->wreplconn->sock->sock;
885         state->wreplconn->sock->sock = NULL;
886         talloc_steal(state, sock);
887
888         /* 
889          * steal the packet_context
890          * note the request DATA_BLOB we just send on the
891          * wrepl_socket (client connection) is still unter the 
892          * packet context and will be send to the wire
893          */
894         packet = state->wreplconn->sock->packet;
895         state->wreplconn->sock->packet = NULL;
896         talloc_steal(state, packet);
897
898         /*
899          * get the fde_flags of the old fde event,
900          * so that we can later set the same flags to the new one
901          */
902         fde_flags = event_get_fd_flags(state->wreplconn->sock->event.fde);
903
904         /*
905          * free the wrepl_socket (client connection)
906          */
907         talloc_free(state->wreplconn->sock);
908         state->wreplconn->sock = NULL;
909
910         /*
911          * now create a wreplsrv_in_connection,
912          * on which we act as server
913          *
914          * NOTE: sock and packet will be stolen by
915          *       wreplsrv_in_connection_merge()
916          */
917         status = wreplsrv_in_connection_merge(state->io->in.partner,
918                                               sock, packet, &wrepl_in);
919         NT_STATUS_NOT_OK_RETURN(status);
920
921         event_set_fd_flags(wrepl_in->conn->event.fde, fde_flags);
922
923         wrepl_in->assoc_ctx.peer_ctx    = state->wreplconn->assoc_ctx.peer_ctx;
924         wrepl_in->assoc_ctx.our_ctx     = 0;
925
926         /* now we can free the wreplsrv_out_connection */
927         talloc_free(state->wreplconn);
928         state->wreplconn = NULL;
929
930         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
931
932         return NT_STATUS_OK;
933 }
934
935 static NTSTATUS wreplsrv_push_notify_inform(struct wreplsrv_push_notify_state *state)
936 {
937         struct wreplsrv_service *service = state->io->in.partner->service;
938         struct wrepl_packet *req = &state->req_packet;
939         struct wrepl_replication *repl_out = &state->req_packet.message.replication;
940         struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
941         NTSTATUS status;
942
943         req->opcode     = WREPL_OPCODE_BITS;
944         req->assoc_ctx  = state->wreplconn->assoc_ctx.peer_ctx;
945         req->mess_type  = WREPL_REPLICATION;
946
947         repl_out->command = state->command;
948
949         status = wreplsrv_fill_wrepl_table(service, state, table_out,
950                                            service->wins_db->local_owner, state->full_table);
951         NT_STATUS_NOT_OK_RETURN(status);
952
953         /* we won't get a reply to a inform message */
954         state->ctrl.send_only           = true;
955
956         state->req = wrepl_request_send(state->wreplconn->sock, req, &state->ctrl);
957         NT_STATUS_HAVE_NO_MEMORY(state->req);
958
959         state->req->async.fn            = wreplsrv_push_notify_handler_req;
960         state->req->async.private       = state;
961
962         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM;
963
964         return NT_STATUS_OK;
965 }
966
967 static NTSTATUS wreplsrv_push_notify_wait_connect(struct wreplsrv_push_notify_state *state)
968 {
969         NTSTATUS status;
970
971         status = wreplsrv_out_connect_recv(state->creq, state, &state->wreplconn);
972         NT_STATUS_NOT_OK_RETURN(status);
973
974         /* is the peer doesn't support inform fallback to update */
975         switch (state->command) {
976         case WREPL_REPL_INFORM:
977                 if (state->wreplconn->assoc_ctx.peer_major < 5) {
978                         state->command = WREPL_REPL_UPDATE;
979                 }
980                 break;
981         case WREPL_REPL_INFORM2:
982                 if (state->wreplconn->assoc_ctx.peer_major < 5) {
983                         state->command = WREPL_REPL_UPDATE2;
984                 }
985                 break;
986         default:
987                 break;
988         }
989
990         switch (state->command) {
991         case WREPL_REPL_UPDATE:
992                 state->full_table = true;
993                 return wreplsrv_push_notify_update(state);
994         case WREPL_REPL_UPDATE2:
995                 state->full_table = false;
996                 return wreplsrv_push_notify_update(state);
997         case WREPL_REPL_INFORM:
998                 state->full_table = true;
999                 return wreplsrv_push_notify_inform(state);
1000         case WREPL_REPL_INFORM2:
1001                 state->full_table = false;
1002                 return wreplsrv_push_notify_inform(state);
1003         default:
1004                 return NT_STATUS_INTERNAL_ERROR;
1005         }
1006
1007         return NT_STATUS_INTERNAL_ERROR;
1008 }
1009
1010 static NTSTATUS wreplsrv_push_notify_wait_inform(struct wreplsrv_push_notify_state *state)
1011 {
1012         NTSTATUS status;
1013
1014         status =  wrepl_request_recv(state->req, state, NULL);
1015         NT_STATUS_NOT_OK_RETURN(status);
1016
1017         state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
1018         return status;
1019 }
1020
1021 static void wreplsrv_push_notify_handler(struct wreplsrv_push_notify_state *state)
1022 {
1023         struct composite_context *c = state->c;
1024
1025         switch (state->stage) {
1026         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
1027                 c->status = wreplsrv_push_notify_wait_connect(state);
1028                 break;
1029         case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM:
1030                 c->status = wreplsrv_push_notify_wait_inform(state);
1031                 break;
1032         case WREPLSRV_PUSH_NOTIFY_STAGE_DONE:
1033                 c->status = NT_STATUS_INTERNAL_ERROR;
1034         }
1035
1036         if (state->stage == WREPLSRV_PUSH_NOTIFY_STAGE_DONE) {
1037                 c->state  = COMPOSITE_STATE_DONE;
1038         }
1039
1040         if (!NT_STATUS_IS_OK(c->status)) {
1041                 c->state = COMPOSITE_STATE_ERROR;
1042         }
1043
1044         if (c->state >= COMPOSITE_STATE_DONE && c->async.fn) {
1045                 c->async.fn(c);
1046         }
1047 }
1048
1049 static void wreplsrv_push_notify_handler_creq(struct composite_context *creq)
1050 {
1051         struct wreplsrv_push_notify_state *state = talloc_get_type(creq->async.private_data,
1052                                                    struct wreplsrv_push_notify_state);
1053         wreplsrv_push_notify_handler(state);
1054         return;
1055 }
1056
1057 static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
1058 {
1059         struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private,
1060                                                    struct wreplsrv_push_notify_state);
1061         wreplsrv_push_notify_handler(state);
1062         return;
1063 }
1064
1065 struct composite_context *wreplsrv_push_notify_send(TALLOC_CTX *mem_ctx, struct wreplsrv_push_notify_io *io)
1066 {
1067         struct composite_context *c = NULL;
1068         struct wreplsrv_service *service = io->in.partner->service;
1069         struct wreplsrv_push_notify_state *state = NULL;
1070         enum winsrepl_partner_type partner_type;
1071
1072         c = talloc_zero(mem_ctx, struct composite_context);
1073         if (!c) goto failed;
1074
1075         state = talloc_zero(c, struct wreplsrv_push_notify_state);
1076         if (!state) goto failed;
1077         state->c        = c;
1078         state->io       = io;
1079
1080         if (io->in.inform) {
1081                 /* we can cache the connection in partner->push->wreplconn */
1082                 partner_type = WINSREPL_PARTNER_PUSH;
1083                 if (io->in.propagate) {
1084                         state->command  = WREPL_REPL_INFORM2;
1085                 } else {
1086                         state->command  = WREPL_REPL_INFORM;
1087                 }
1088         } else {
1089                 /* we can NOT cache the connection */
1090                 partner_type = WINSREPL_PARTNER_NONE;
1091                 if (io->in.propagate) {
1092                         state->command  = WREPL_REPL_UPDATE2;
1093                 } else {
1094                         state->command  = WREPL_REPL_UPDATE;
1095                 }       
1096         }
1097
1098         c->state        = COMPOSITE_STATE_IN_PROGRESS;
1099         c->event_ctx    = service->task->event_ctx;
1100         c->private_data = state;
1101
1102         state->stage    = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT;
1103         state->creq     = wreplsrv_out_connect_send(io->in.partner, partner_type, NULL);
1104         if (!state->creq) goto failed;
1105
1106         state->creq->async.fn           = wreplsrv_push_notify_handler_creq;
1107         state->creq->async.private_data = state;
1108
1109         return c;
1110 failed:
1111         talloc_free(c);
1112         return NULL;
1113 }
1114
1115 NTSTATUS wreplsrv_push_notify_recv(struct composite_context *c)
1116 {
1117         NTSTATUS status;
1118
1119         status = composite_wait(c);
1120
1121         talloc_free(c);
1122         return status;
1123 }